diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolver.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolver.java new file mode 100644 index 000000000..91d10de48 --- /dev/null +++ b/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}. + * + *

{@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 tablePath) throws SQLException; +} diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolverProvider.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolverProvider.java new file mode 100644 index 000000000..48d8e1d65 --- /dev/null +++ b/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. + * + *

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; + } +} diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolvers.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolvers.java new file mode 100644 index 000000000..b17ef847d --- /dev/null +++ b/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.")); + } +} diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DirectDeploymentContext.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DirectDeploymentContext.java new file mode 100644 index 000000000..8b0795fc2 --- /dev/null +++ b/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}: + * + *

+ * + *

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); + } +} diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDriver.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDriver.java index c52edf11e..b1067ef4f 100644 --- a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDriver.java +++ b/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: + * + *

*/ 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; } diff --git a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/TableService.java b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/TableService.java new file mode 100644 index 000000000..92a6283cf --- /dev/null +++ b/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. + * + *

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. + * + *

These entry points are connection-free: 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> logHooks, List path, Schema avroSchema, Map 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 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 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; + } + } +} diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolverProviderTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolverProviderTest.java new file mode 100644 index 000000000..fb78cdb06 --- /dev/null +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolverProviderTest.java @@ -0,0 +1,17 @@ +package com.linkedin.hoptimator.jdbc; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + + +class DatabaseConfigResolverProviderTest { + + @Test + void priorityDefaultsToZero() { + // A provider that does not override priority() (e.g. the jdbc test fixture) reports the default. + DatabaseConfigResolverProvider provider = new TestDatabaseConfigResolverProvider(); + + assertThat(provider.priority()).isEqualTo(0); + } +} diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolversTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolversTest.java new file mode 100644 index 000000000..f88a0b176 --- /dev/null +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolversTest.java @@ -0,0 +1,23 @@ +package com.linkedin.hoptimator.jdbc; + +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; +import java.util.List; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; + + +class DatabaseConfigResolversTest { + + @Test + void forPropertiesReturnsServiceLoadedResolver() throws SQLException { + // The jdbc test classpath registers TestDatabaseConfigResolverProvider (priority 0), which + // resolves the database identifier to the path's schema segment. + DatabaseConfigResolver resolver = DatabaseConfigResolvers.forProperties(new Properties()); + + assertThat(resolver).isNotNull(); + assertThat(resolver.databaseName(List.of("MYDB", "myTable"))).isEqualTo("MYDB"); + } +} diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DirectDeploymentContextTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DirectDeploymentContextTest.java new file mode 100644 index 000000000..2fb845865 --- /dev/null +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/DirectDeploymentContextTest.java @@ -0,0 +1,75 @@ +package com.linkedin.hoptimator.jdbc; + +import org.apache.avro.Schema; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; +import java.util.List; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +class DirectDeploymentContextTest { + + private static Schema avroSchema() { + return new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"R\"," + + "\"namespace\":\"com.example\",\"fields\":[{\"name\":\"ID\",\"type\":\"int\"}]}"); + } + + private static DatabaseConfigResolver resolverReturning(@Nullable Properties props) { + return new DatabaseConfigResolver() { + @Override + public @Nullable Properties databaseProperties(@Nullable String catalog, @Nullable String schema, + String connectionPrefix) { + return props; + } + + @Override + public String databaseName(List tablePath) { + return tablePath.get(tablePath.size() - 2); + } + }; + } + + @Test + void propertiesReturnsSuppliedBag() { + Properties props = new Properties(); + props.setProperty("k8s.namespace", "ns"); + DirectDeploymentContext context = new DirectDeploymentContext(props, resolverReturning(null), avroSchema()); + + assertThat(context.properties()).isSameAs(props); + } + + @Test + void rowTypeDerivedFromAvroSchema() { + Schema avroSchema = avroSchema(); + DirectDeploymentContext context = + new DirectDeploymentContext(new Properties(), resolverReturning(null), avroSchema); + + assertThat(context.avroSchema()).isSameAs(avroSchema); + assertThat(context.rowType().isStruct()).isTrue(); + assertThat(context.rowType().getFieldNames()).contains("ID"); + } + + @Test + void rowTypeThrowsWhenAbsent() { + DirectDeploymentContext context = + new DirectDeploymentContext(new Properties(), resolverReturning(null)); + + assertThatThrownBy(context::rowType) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("No Avro schema is carried"); + } + + @Test + void databasePropertiesDelegatesToResolver() { + Properties dbProps = new Properties(); + dbProps.setProperty("bootstrap.servers", "localhost:9092"); + DirectDeploymentContext context = + new DirectDeploymentContext(new Properties(), resolverReturning(dbProps)); + + assertThat(context.databaseProperties(null, "KAFKA", "jdbc:kafka://")).isSameAs(dbProps); + } +} diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/HoptimatorDriverTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/HoptimatorDriverTest.java index 0df50e4c0..5c8120fa6 100644 --- a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/HoptimatorDriverTest.java +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/HoptimatorDriverTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -139,6 +140,18 @@ void testRowTypeThrowsForMissingTable() throws SQLException { } } + @Test + void testRowTypeDerivedFromAvroSchemaOnDirectContext() throws SQLException { + Schema avro = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"R\"," + + "\"namespace\":\"com.example\",\"fields\":[{\"name\":\"ID\",\"type\":\"int\"}]}"); + DirectDeploymentContext context = new DirectDeploymentContext(new Properties(), null, avro); + Source source = new Source("KAFKA", Arrays.asList("KAFKA", "my_topic"), Collections.emptyMap()); + + RelDataType rowType = HoptimatorDriver.rowType(source, context); + assertTrue(rowType.isStruct()); + assertTrue(rowType.getFieldNames().contains("ID")); + } + @Test void testRowTypeThrowsForUnknownContextType() { Source source = new Source("KAFKA", Arrays.asList("KAFKA", "my_topic"), Collections.emptyMap()); @@ -157,6 +170,52 @@ public Properties databaseProperties(String catalog, String schema, String conne assertThrows(SQLException.class, () -> HoptimatorDriver.rowType(source, unknown)); } + @Test + void testValueSchemaReturnsNullForSchemaFreeDirectContext() { + DirectDeploymentContext context = new DirectDeploymentContext(new Properties(), null, null); + Source source = new Source("KAFKA", Arrays.asList("KAFKA", "my_topic"), Collections.emptyMap()); + + assertNull(HoptimatorDriver.valueSchema(source, context)); + } + + @Test + void testValueSchemaSplitsValueFromCarriedMergedAvroOnDirectContext() { + // The caller's carried Avro is a merged key+value record (KEY_-prefixed key fields). valueSchema + // must return only the value portion (keys handled separately via key.fields) and preserve a + // nested record's namespace losslessly rather than re-synthesizing from the flat row type. + Schema merged = new Schema.Parser().parse("{" + + "\"type\":\"record\",\"name\":\"StoreValue\",\"namespace\":\"com.example.kafka\",\"fields\":[" + + "{\"name\":\"KEY_id\",\"type\":\"int\"}," + + "{\"name\":\"widget\",\"type\":{\"type\":\"record\",\"name\":\"Widget\"," + + "\"namespace\":\"com.example.custom\",\"fields\":[{\"name\":\"w\",\"type\":\"string\"}]}}" + + "]}"); + DirectDeploymentContext context = new DirectDeploymentContext(new Properties(), null, merged); + Source source = new Source("KAFKA", Arrays.asList("KAFKA", "my_topic"), Collections.emptyMap()); + + Schema value = HoptimatorDriver.valueSchema(source, context); + + assertNull(value.getField("KEY_id"), "key field excluded from value schema"); + assertNotNull(value.getField("widget"), "value field retained"); + assertEquals("com.example.custom", value.getField("widget").schema().getNamespace(), + "nested namespace preserved losslessly"); + } + + @Test + void testProvidedAvroSchemaReturnsCarriedSchemaFromDirectContext() { + Schema avro = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"R\"," + + "\"namespace\":\"com.example\",\"fields\":[{\"name\":\"ID\",\"type\":\"int\"}]}"); + DirectDeploymentContext context = new DirectDeploymentContext(new Properties(), null, avro); + + assertSame(avro, HoptimatorDriver.providedAvroSchema(context)); + } + + @Test + void testProvidedAvroSchemaReturnsNullWhenNoneCarried() { + DirectDeploymentContext context = new DirectDeploymentContext(new Properties(), null, null); + + assertNull(HoptimatorDriver.providedAvroSchema(context)); + } + @Test void testValueSchemaReturnsNullForTableWithoutNativeAvro() throws SQLException { try (HoptimatorConnection connection = diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TableServiceTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TableServiceTest.java new file mode 100644 index 000000000..c7e154a3c --- /dev/null +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TableServiceTest.java @@ -0,0 +1,202 @@ +package com.linkedin.hoptimator.jdbc; + +import com.linkedin.hoptimator.Deployer; +import com.linkedin.hoptimator.DeploymentContext; +import com.linkedin.hoptimator.Source; +import com.linkedin.hoptimator.util.DeploymentService; +import org.apache.avro.Schema; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.sql.SQLException; +import java.util.Collections; +import java.util.List; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + + +/** + * Unit tests for the SQL-free direct {@link TableService} API. + */ +@ExtendWith(MockitoExtension.class) +class TableServiceTest { + + private static final String RECORD_SCHEMA = "{" + + "\"type\":\"record\",\"name\":\"MyTable\",\"namespace\":\"com.linkedin.test\"," + + "\"fields\":[" + + "{\"name\":\"id\",\"type\":\"long\"}," + + "{\"name\":\"name\",\"type\":\"string\"}" + + "]}"; + + private final List path = List.of("UTIL", "myTable"); + + @Mock + private MockedStatic resolvers; + + @Mock + private MockedStatic deployment; + + @Test + void createDryRunDerivesRowTypeFromAvroSchema() throws SQLException { + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + HoptimatorDdlUtils.SpecifyResult result = + TableService.create(new Properties(), Collections.emptyList(), path, recordSchema(), + Collections.emptyMap(), false, true); + + assertThat(result).isNotNull(); + assertThat(result.viewPath).endsWith("myTable"); + + RelDataType rowType = result.sinkRowType; + assertThat(rowType.isStruct()).isTrue(); + assertThat(rowType.getFieldNames()).containsExactly("id", "name"); + assertThat(rowType.getField("id", false, false).getType().getSqlTypeName()).isEqualTo(SqlTypeName.BIGINT); + assertThat(rowType.getField("name", false, false).getType().getSqlTypeName()).isEqualTo(SqlTypeName.VARCHAR); + } + + @Test + void createRejectsPathWithoutDatabaseAndTable() { + assertThatThrownBy(() -> + TableService.create(new Properties(), Collections.emptyList(), List.of("onlyOne"), recordSchema(), + Collections.emptyMap(), false, true)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("database and a table name"); + } + + @Test + void createRejectsNullSchema() { + assertThatThrownBy(() -> + TableService.create(new Properties(), Collections.emptyList(), path, null, + Collections.emptyMap(), false, true)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Avro schema is required"); + } + + @Test + void rejectsNonRecordSchema() { + Schema primitive = Schema.create(Schema.Type.STRING); + assertThatThrownBy(() -> + TableService.create(new Properties(), Collections.emptyList(), path, primitive, + Collections.emptyMap(), false, true)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("must be a record"); + } + + @Test + void deleteRejectsPathWithoutDatabaseAndTable() { + assertThatThrownBy(() -> TableService.delete(new Properties(), List.of("onlyOne"))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("database and a table name"); + } + + @Test + void deleteRunsValidationAndDeployerTeardown() throws SQLException { + Deployer deployer = mock(Deployer.class); + List deployers = Collections.singletonList(deployer); + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class))) + .thenReturn(deployers); + + TableService.delete(new Properties(), path); + + deployment.verify(() -> DeploymentService.delete(deployers), times(1)); + deployment.verify(() -> DeploymentService.restore(any()), never()); + } + + @Test + void deleteRestoresAndRethrowsWhenTeardownFails() { + Deployer deployer = mock(Deployer.class); + List deployers = Collections.singletonList(deployer); + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class))) + .thenReturn(deployers); + deployment.when(() -> DeploymentService.delete(deployers)) + .thenThrow(new SQLException("teardown boom")); + + assertThatThrownBy(() -> TableService.delete(new Properties(), path)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("teardown boom"); + + deployment.verify(() -> DeploymentService.restore(deployers), times(1)); + } + + @Test + void createFailsWhenTableAlreadyExists() throws SQLException { + // Direct-path CREATE (updateIfExists=false) against an existing table must fail, mirroring the + // SQL path's "already exists, use OR REPLACE" instead of silently skipping per deployer. + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + Deployer deployer = mock(Deployer.class); + when(deployer.exists()).thenReturn(true); + deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class))) + .thenReturn(Collections.singletonList(deployer)); + + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), path, + recordSchema(), Collections.emptyMap(), false, false)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("already exists"); + + deployment.verify(() -> DeploymentService.create(any()), never()); + } + + @Test + void createProceedsWhenTableDoesNotExist() throws SQLException { + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + Deployer deployer = mock(Deployer.class); + when(deployer.exists()).thenReturn(false); + List deployers = Collections.singletonList(deployer); + deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class))) + .thenReturn(deployers); + + TableService.create(new Properties(), Collections.emptyList(), path, recordSchema(), + Collections.emptyMap(), false, false); + + deployment.verify(() -> DeploymentService.create(deployers), times(1)); + } + + @Test + void updateIfExistsBypassesTheGuardAndDoesNotConsultExists() throws SQLException { + DatabaseConfigResolver resolver = stubResolver(); + resolvers.when(() -> DatabaseConfigResolvers.forProperties(any())).thenReturn(resolver); + Deployer deployer = mock(Deployer.class); + List deployers = Collections.singletonList(deployer); + deployment.when(() -> DeploymentService.deployers(any(Source.class), any(DeploymentContext.class))) + .thenReturn(deployers); + + TableService.create(new Properties(), Collections.emptyList(), path, recordSchema(), + Collections.emptyMap(), true, false); + + deployment.verify(() -> DeploymentService.update(deployers), times(1)); + verify(deployer, never()).exists(); + } + + private static Schema recordSchema() { + return new Schema.Parser().parse(RECORD_SCHEMA); + } + + private DatabaseConfigResolver stubResolver() { + DatabaseConfigResolver resolver = mock(DatabaseConfigResolver.class); + try { + when(resolver.databaseName(path)).thenReturn("test-database"); + } catch (SQLException e) { + throw new RuntimeException(e); + } + return resolver; + } +} diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestDatabaseConfigResolverProvider.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestDatabaseConfigResolverProvider.java new file mode 100644 index 000000000..8434aa84f --- /dev/null +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/TestDatabaseConfigResolverProvider.java @@ -0,0 +1,33 @@ +package com.linkedin.hoptimator.jdbc; + +import java.util.List; +import java.util.Properties; + +import javax.annotation.Nullable; + + +/** + * Minimal registry-native {@link DatabaseConfigResolverProvider} for {@code hoptimator-jdbc} unit + * tests, which run without a real registry (e.g. hoptimator-k8s) on the classpath. It resolves the + * database identifier to the path's schema segment and reports no per-{@code Database} config, which + * is enough to exercise {@link TableService}'s Avro handling, guards, and dry-run rendering without + * standing up a backend. + */ +public final class TestDatabaseConfigResolverProvider implements DatabaseConfigResolverProvider { + + @Override + public DatabaseConfigResolver resolver(Properties connectionProperties) { + return new DatabaseConfigResolver() { + @Override + public @Nullable Properties databaseProperties(@Nullable String catalog, @Nullable String schema, + String connectionPrefix) { + return null; + } + + @Override + public String databaseName(List tablePath) { + return tablePath.get(tablePath.size() - 2); + } + }; + } +} diff --git a/hoptimator-jdbc/src/test/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider b/hoptimator-jdbc/src/test/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider new file mode 100644 index 000000000..490b1d54e --- /dev/null +++ b/hoptimator-jdbc/src/test/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider @@ -0,0 +1 @@ +com.linkedin.hoptimator.jdbc.TestDatabaseConfigResolverProvider diff --git a/hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/CatalogResolver.java b/hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/CatalogResolver.java new file mode 100644 index 000000000..71931704b --- /dev/null +++ b/hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/CatalogResolver.java @@ -0,0 +1,61 @@ +package com.linkedin.hoptimator.jdbc; + +import java.sql.DriverManager; +import java.sql.SQLException; +import java.util.List; + +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.Table; + + +/** + * Resolves a table's schema from the live {@code k8s} catalog — the programmatic equivalent of + * quidem's {@code !describe}. Used by the {@code *TableServiceIntegrationTest}s to verify that a + * table created via the SQL-free {@link TableService} really registered (and, after delete, really + * went away), independently of what {@code create} returned. + * + *

Each call opens a fresh connection so the catalog is re-read (Calcite caches per connection), + * and navigation follows the table path: {@code [SCHEMA, TABLE]} or {@code [CATALOG, SCHEMA, TABLE]}. + * Note: this reliably reflects creates, but not deletes — Calcite's + * {@code JdbcSchema}/{@code ClusterSchema} caches table existence at a level shared across + * connections, so a dropped table can still resolve. Use it to verify registration, not removal. + */ +public final class CatalogResolver { + + private CatalogResolver() { + } + + /** Resolves the row type for {@code path}, or {@code null} if the table is not present. */ + private static RelDataType resolveRowType(List path) { + try (HoptimatorConnection conn = + (HoptimatorConnection) DriverManager.getConnection("jdbc:hoptimator://catalogs=k8s")) { + SchemaPlus schema = conn.calciteConnection().getRootSchema(); + for (String segment : path.subList(0, path.size() - 1)) { + if (schema == null) { + return null; + } + schema = schema.subSchemas().get(segment); + } + if (schema == null) { + return null; + } + Table table = schema.tables().get(path.get(path.size() - 1)); + return table == null ? null : table.getRowType(conn.calciteConnection().getTypeFactory()); + } catch (SQLException e) { + throw new RuntimeException(e); + } + } + + /** + * Resolves {@code path} immediately, failing if it is not present. The store deployers run + * synchronously, so a table must be resolvable right after {@code create} returns + */ + public static RelDataType resolve(List path) throws SQLException { + RelDataType rowType = resolveRowType(path); + if (rowType == null) { + throw new SQLException("Table not resolvable after create: " + path); + } + return rowType; + } +} diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolver.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolver.java new file mode 100644 index 000000000..b39ff6a9a --- /dev/null +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolver.java @@ -0,0 +1,134 @@ +package com.linkedin.hoptimator.k8s; + +import com.linkedin.hoptimator.jdbc.DatabaseConfigResolver; +import com.linkedin.hoptimator.k8s.models.V1alpha1Database; +import org.apache.calcite.avatica.ConnectStringParser; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + + +/** + * A {@link DatabaseConfigResolver} that reads {@code Database} config directly from K8s + * {@code Database} CRDs — no Calcite catalog, no {@code java.sql.Connection}. It reconstructs the + * same effective JDBC URL that {@link K8sDatabaseTable} would build for the catalog, then parses it + * into connection properties, so the connection-free direct path resolves config identically to the + * SQL path. + */ +public final class K8sDatabaseConfigResolver implements DatabaseConfigResolver { + + private static final Logger LOG = LoggerFactory.getLogger(K8sDatabaseConfigResolver.class); + + private final Properties connectionProperties; + private final K8sContext context; + private List cachedDatabases; + + public K8sDatabaseConfigResolver(Properties connectionProperties) { + this(connectionProperties, K8sContext.create(connectionProperties)); + } + + /** Test seam: inject a (mockable) {@link K8sContext} instead of building one from the properties. */ + K8sDatabaseConfigResolver(Properties connectionProperties, K8sContext context) { + this.connectionProperties = connectionProperties; + this.context = context; + } + + @Override + public @Nullable Properties databaseProperties(@Nullable String catalog, @Nullable String schema, + String connectionPrefix) { + if (catalog == null && schema == null) { + return null; + } + K8sDatabaseTable.Row row; + try { + row = findDatabase(catalog, schema); + } catch (SQLException e) { + // Fail loudly rather than returning null: the deployer providers treat a null here as "no + // config for this store" and deploy nothing, which would report success while a K8s error + // meant we never figured out what to deploy. The SPI signature can't throw, so wrap it. + throw new IllegalStateException("Failed to resolve Database config for " + + (catalog != null ? catalog : schema) + ": " + e.getMessage(), e); + } + if (row == null || row.URL == null || !row.URL.startsWith(connectionPrefix)) { + return null; + } + String joined = K8sDatabaseTable.joinedUrl(row, connectionProperties); + Properties properties = new Properties(); + try { + properties.putAll(ConnectStringParser.parse(joined.substring(connectionPrefix.length()))); + } catch (SQLException e) { + LOG.debug("Could not parse URL for schema '{}': {}", schema, e.getMessage()); + return null; + } + return properties; + } + + @Override + public String databaseName(List tablePath) throws SQLException { + String schema = tablePath.get(tablePath.size() - 2); + String catalog = tablePath.size() >= 3 ? tablePath.get(tablePath.size() - 3) : null; + K8sDatabaseTable.Row row = findDatabase(catalog, schema); + if (row == null) { + throw new SQLException("No Database is registered for " + + (catalog != null ? catalog + "." + schema : schema) + "."); + } + // The database identifier is always the Database CRD name, for both schema- and catalog-style + // Databases — matching the SQL path, which injects it into the JDBC URL as database= (see + // K8sDatabaseTable#joinedUrl). It names deployed resources and matches Table/Job template + // `databases` filters; the store-level schema is carried separately by Source#schema(). + return row.NAME; + } + + private @Nullable K8sDatabaseTable.Row findDatabase(@Nullable String catalog, @Nullable String schema) + throws SQLException { + for (K8sDatabaseTable.Row row : listDatabases()) { + if (matches(row, catalog, schema)) { + return row; + } + } + return null; + } + + /** + * The Database CRDs, listed once and cached for this resolver's lifetime. A resolver is built once + * per direct-path operation ({@code TableService.create}/{@code delete}), which may resolve + * several databases (e.g. a logical table's tiers each hit this), so listing once per operation + * avoids repeated K8s round-trips. Deliberately instance-scoped rather than static/global: a fresh + * resolver per operation still observes newly created/deleted Databases. + * + *

TODO: This lists all Database CRDs and filters client-side ({@link #matches}) because the K8s + * API cannot field-select on {@code spec.catalog}/{@code spec.schema} — only {@code metadata.name} + * (via {@code K8sApi#get}) or labels (via {@code K8sApi#select(labelSelector)}). If Databases were + * labelled with their catalog/schema at creation, the filter could be pushed server-side. Left as + * list-and-filter for now: cardinality is low (one CRD per database) and it mirrors how + * {@link K8sDatabaseTable} (the SQL/catalog path) enumerates Databases. + */ + private List listDatabases() throws SQLException { + if (cachedDatabases == null) { + List rows = new ArrayList<>(); + // No catch: a failed list must surface (see databaseName/databaseProperties) rather than be + // swallowed into an empty list that reads as "no Databases". cachedDatabases stays null on + // failure, so it is not cached and a transient error can recover on the next call. + for (V1alpha1Database db : new K8sApi<>(context, K8sApiEndpoints.DATABASES).list()) { + rows.add(K8sDatabaseTable.rowOf(db)); + } + cachedDatabases = rows; + } + return cachedDatabases; + } + + private static boolean matches(K8sDatabaseTable.Row row, @Nullable String catalog, @Nullable String schema) { + if (catalog != null) { + // Catalog-style Database (e.g. MYSQL): config lives on the catalog CRD; the requested + // `schema` is a sub-schema that shares this connection. + return catalog.equalsIgnoreCase(row.CATALOG); + } + // Schema-style Database (e.g. KAFKA, VENICE): match by schema name. + return row.CATALOG == null && schema != null && schema.equalsIgnoreCase(K8sDatabaseTable.schemaName(row)); + } +} diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProvider.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProvider.java new file mode 100644 index 000000000..5b4c2ddf7 --- /dev/null +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProvider.java @@ -0,0 +1,21 @@ +package com.linkedin.hoptimator.k8s; + +import com.linkedin.hoptimator.jdbc.DatabaseConfigResolver; +import com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider; + +import java.util.Properties; + + +/** Supplies a {@link K8sDatabaseConfigResolver} so the direct path resolves config from CRDs. */ +public class K8sDatabaseConfigResolverProvider implements DatabaseConfigResolverProvider { + + @Override + public DatabaseConfigResolver resolver(Properties connectionProperties) { + return new K8sDatabaseConfigResolver(connectionProperties); + } + + @Override + public int priority() { + return 1; + } +} diff --git a/hoptimator-k8s/src/main/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider b/hoptimator-k8s/src/main/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider new file mode 100644 index 000000000..06872f5ea --- /dev/null +++ b/hoptimator-k8s/src/main/resources/META-INF/services/com.linkedin.hoptimator.jdbc.DatabaseConfigResolverProvider @@ -0,0 +1 @@ +com.linkedin.hoptimator.k8s.K8sDatabaseConfigResolverProvider diff --git a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProviderTest.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProviderTest.java new file mode 100644 index 000000000..f910dfece --- /dev/null +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProviderTest.java @@ -0,0 +1,33 @@ +package com.linkedin.hoptimator.k8s; + +import com.linkedin.hoptimator.jdbc.DatabaseConfigResolver; +import org.junit.jupiter.api.Test; + +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; + + +class K8sDatabaseConfigResolverProviderTest { + + private final K8sDatabaseConfigResolverProvider provider = new K8sDatabaseConfigResolverProvider(); + + @Test + void resolverReturnsK8sResolver() { + // Offline connection properties: server+token make K8sContext build an ApiClient via + // Config.fromToken (no kubeconfig read, no network), so constructing the resolver does not + // require an ambient cluster. The test only asserts the resolver type. + Properties props = new Properties(); + props.setProperty("k8s.server", "https://localhost:1"); + props.setProperty("k8s.token", "test-token"); + props.setProperty("k8s.namespace", "default"); + DatabaseConfigResolver resolver = provider.resolver(props); + + assertThat(resolver).isInstanceOf(K8sDatabaseConfigResolver.class); + } + + @Test + void priorityIsOne() { + assertThat(provider.priority()).isEqualTo(1); + } +} diff --git a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverTest.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverTest.java new file mode 100644 index 000000000..aa9aebe12 --- /dev/null +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverTest.java @@ -0,0 +1,178 @@ +package com.linkedin.hoptimator.k8s; + +import com.linkedin.hoptimator.k8s.models.V1alpha1Database; +import com.linkedin.hoptimator.k8s.models.V1alpha1DatabaseList; +import com.linkedin.hoptimator.k8s.models.V1alpha1DatabaseSpec; +import io.kubernetes.client.openapi.ApiException; +import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.util.generic.GenericKubernetesApi; +import io.kubernetes.client.util.generic.KubernetesApiResponse; +import io.kubernetes.client.util.generic.options.ListOptions; +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + + +class K8sDatabaseConfigResolverTest { + + private static final String NAMESPACE = "test-ns"; + + /** A K8sContext whose {@code Database} list returns {@code dbs} (Databases are namespaced). */ + private static K8sContext contextReturning(V1alpha1Database... dbs) { + V1alpha1DatabaseList list = new V1alpha1DatabaseList(); + list.setItems(Arrays.asList(dbs)); + KubernetesApiResponse resp = mockResponse(); + when(resp.getObject()).thenReturn(list); + return contextListing(resp); + } + + /** A K8sContext whose {@code Database} list fails with an API error (non-transient status). */ + private static K8sContext contextWithListFailure() { + KubernetesApiResponse resp = mockResponse(); + when(resp.getHttpStatusCode()).thenReturn(500); + try { + doThrow(new ApiException("boom")).when(resp).throwsApiException(); + } catch (ApiException e) { + throw new RuntimeException(e); + } + return contextListing(resp); + } + + @SuppressWarnings("unchecked") + private static KubernetesApiResponse mockResponse() { + return mock(KubernetesApiResponse.class); + } + + private static K8sContext contextListing(KubernetesApiResponse resp) { + K8sContext context = mock(K8sContext.class); + @SuppressWarnings("unchecked") + GenericKubernetesApi generic = mock(GenericKubernetesApi.class); + when(context.namespace()).thenReturn(NAMESPACE); + when(context.generic(K8sApiEndpoints.DATABASES)).thenReturn(generic); + when(generic.list(eq(NAMESPACE), any(ListOptions.class))).thenReturn(resp); + return context; + } + + private static K8sDatabaseConfigResolver resolver(V1alpha1Database... dbs) { + return new K8sDatabaseConfigResolver(new Properties(), contextReturning(dbs)); + } + + private static V1alpha1Database db(String name, String url, String catalog, String schema) { + return new V1alpha1Database() + .metadata(new V1ObjectMeta().name(name)) + .spec(new V1alpha1DatabaseSpec().url(url).catalog(catalog).schema(schema)); + } + + @Test + void databasePropertiesReturnsNullWhenNoCatalogOrSchema() { + assertThat(resolver().databaseProperties(null, null, "jdbc:kafka://")).isNull(); + } + + @Test + void databasePropertiesParsesUrlForSchemaStyleDatabase() { + V1alpha1Database kafka = db("kafka-database", + "jdbc:kafka://bootstrap.servers=localhost:9092", null, "KAFKA"); + + Properties props = resolver(kafka).databaseProperties(null, "KAFKA", "jdbc:kafka://"); + + assertThat(props).isNotNull(); + assertThat(props.getProperty("bootstrap.servers")).isEqualTo("localhost:9092"); + // joinedUrl injects the CRD name as database=. + assertThat(props.getProperty("database")).isEqualTo("kafka-database"); + } + + @Test + void databasePropertiesReturnsNullWhenUrlDoesNotMatchPrefix() { + V1alpha1Database venice = db("venice", "jdbc:venice://clusters=venice-cluster0", null, "VENICE"); + + // Database exists for VENICE but the requested prefix is a different store type. + assertThat(resolver(venice).databaseProperties(null, "VENICE", "jdbc:kafka://")).isNull(); + } + + @Test + void databasePropertiesReturnsNullWhenNoDatabaseMatches() { + assertThat(resolver().databaseProperties(null, "MISSING", "jdbc:kafka://")).isNull(); + } + + @Test + void databaseNameReturnsCrdNameForSchemaStyle() throws Exception { + V1alpha1Database kafka = db("kafka-database", + "jdbc:kafka://bootstrap.servers=localhost:9092", null, "KAFKA"); + + assertThat(resolver(kafka).databaseName(Arrays.asList("KAFKA", "my_topic"))).isEqualTo("kafka-database"); + } + + @Test + void databaseNameMatchesCatalogStyleByCatalog() throws Exception { + V1alpha1Database mysql = db("mysql", "jdbc:mysql-hoptimator://url=jdbc:mysql://localhost:3306", + "MYSQL", null); + + // Catalog-style: three-segment path [CATALOG, SCHEMA, TABLE] matches on the catalog CRD. + assertThat(resolver(mysql).databaseName(Arrays.asList("MYSQL", "test_database", "orders"))) + .isEqualTo("mysql"); + } + + @Test + void databaseNameThrowsWhenNoDatabaseRegistered() { + assertThatThrownBy(() -> resolver().databaseName(Arrays.asList("UNKNOWN", "t"))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("No Database is registered"); + } + + @Test + void databaseNameThrowsWhenListFails() { + K8sDatabaseConfigResolver resolver = new K8sDatabaseConfigResolver(new Properties(), contextWithListFailure()); + + // A failed list must surface as an error, not be swallowed into "no Database registered". + assertThatThrownBy(() -> resolver.databaseName(Arrays.asList("KAFKA", "t"))) + .isInstanceOf(SQLException.class) + .hasMessageNotContaining("No Database is registered"); + } + + @Test + void databasePropertiesThrowsWhenListFails() { + K8sDatabaseConfigResolver resolver = new K8sDatabaseConfigResolver(new Properties(), contextWithListFailure()); + + // Must fail loudly rather than returning null (which reads as "no config, deploy nothing"). + assertThatThrownBy(() -> resolver.databaseProperties(null, "KAFKA", "jdbc:kafka://")) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Failed to resolve Database config"); + } + + @Test + void listsDatabasesOncePerResolverAcrossMultipleResolutions() throws Exception { + V1alpha1DatabaseList list = new V1alpha1DatabaseList(); + list.setItems(Collections.singletonList( + db("kafka-database", "jdbc:kafka://bootstrap.servers=localhost:9092", null, "KAFKA"))); + KubernetesApiResponse resp = mockResponse(); + when(resp.getObject()).thenReturn(list); + @SuppressWarnings("unchecked") + GenericKubernetesApi generic = mock(GenericKubernetesApi.class); + K8sContext context = mock(K8sContext.class); + when(context.namespace()).thenReturn(NAMESPACE); + when(context.generic(K8sApiEndpoints.DATABASES)).thenReturn(generic); + when(generic.list(eq(NAMESPACE), any(ListOptions.class))).thenReturn(resp); + K8sDatabaseConfigResolver resolver = new K8sDatabaseConfigResolver(new Properties(), context); + + // Several resolutions on the same resolver, as a logical table's tiers would trigger. + resolver.databaseName(Arrays.asList("KAFKA", "t1")); + resolver.databaseProperties(null, "KAFKA", "jdbc:kafka://"); + resolver.databaseName(Arrays.asList("KAFKA", "t2")); + + // The per-resolver cache means the Database CRDs are listed only once. + verify(generic, times(1)).list(eq(NAMESPACE), any(ListOptions.class)); + } +} diff --git a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTableServiceIntegrationTest.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTableServiceIntegrationTest.java new file mode 100644 index 000000000..018d20e85 --- /dev/null +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTableServiceIntegrationTest.java @@ -0,0 +1,67 @@ +package com.linkedin.hoptimator.k8s; + +import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils; +import com.linkedin.hoptimator.jdbc.TableService; +import org.apache.avro.Schema; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Integration tests using the SQL-free {@link TableService} direct API. ADS is a read-only demo ({@code demodb}) + * schema with no dedicated store deployer, so this covers what the direct path supports there — Avro schema + * derivation, dry-run rendering, and error handling — rather than store resolution. + */ +@Tag("integration") +public class K8sTableServiceIntegrationTest { + + @Test + void createTableDerivesRowType() throws SQLException { + Schema schema = record("ts-newtable", nullable("i", Schema.Type.INT), nullable("s", Schema.Type.STRING)); + try { + HoptimatorDdlUtils.SpecifyResult result = TableService.create(new Properties(), Collections.emptyList(), + List.of("ADS", "ts-newtable"), schema, Map.of(), true, false); + assertEquals(List.of("i", "s"), result.sinkRowType.getFieldNames()); + assertEquals(SqlTypeName.INTEGER, result.sinkRowType.getField("i", false, false).getType().getSqlTypeName()); + assertEquals(SqlTypeName.VARCHAR, result.sinkRowType.getField("s", false, false).getType().getSqlTypeName()); + } finally { + TableService.delete(new Properties(), List.of("ADS", "ts-newtable")); + } + } + + @Test + void dryRunDerivesRowTypeWithoutMutation() throws SQLException { + Schema schema = record("ts-dryruntable", nullable("i", Schema.Type.INT)); + HoptimatorDdlUtils.SpecifyResult result = TableService.create(new Properties(), Collections.emptyList(), + List.of("ADS", "ts-dryruntable"), schema, Map.of(), false, true); + assertEquals(List.of("i"), result.sinkRowType.getFieldNames()); + } + + @Test + void createInUnknownDatabaseFails() { + Schema schema = record("t", nullable("i", Schema.Type.INT)); + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), + List.of("NOSUCHDB", "t"), schema, Map.of(), true, false)) + .isInstanceOf(SQLException.class); + } + + private static Schema record(String name, Schema.Field... fields) { + return Schema.createRecord(name.replaceAll("\\W", "_"), null, "com.linkedin.hoptimator.test", false, + Arrays.asList(fields)); + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } +} diff --git a/hoptimator-kafka/src/test/java/com/linkedin/hoptimator/kafka/KafkaTableServiceIntegrationTest.java b/hoptimator-kafka/src/test/java/com/linkedin/hoptimator/kafka/KafkaTableServiceIntegrationTest.java new file mode 100644 index 000000000..4d4ddecf5 --- /dev/null +++ b/hoptimator-kafka/src/test/java/com/linkedin/hoptimator/kafka/KafkaTableServiceIntegrationTest.java @@ -0,0 +1,96 @@ +package com.linkedin.hoptimator.kafka; + +import com.linkedin.hoptimator.jdbc.CatalogResolver; +import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils; +import com.linkedin.hoptimator.jdbc.TableService; +import org.apache.avro.Schema; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Integration tests using the SQL-free {@link TableService}. The lifecycle test covers dry-run, + * real create, schema verification, via {@link CatalogResolver}) and drop (verified by resolving to absent). + */ +@Tag("integration") +public class KafkaTableServiceIntegrationTest { + + private static final String SCHEMA = "KAFKA"; + private final Properties properties = new Properties(); + + @Test + void kafkaCreateTableLifecycle() throws SQLException { + String table = "ts-create-table-test"; + Schema schema = record("KEY", Schema.Type.STRING, "VALUE", Schema.Type.BYTES); + try { + // Dry-run previews the KafkaTopic YAML with the requested partition count, no mutation. + HoptimatorDdlUtils.SpecifyResult preview = TableService.create(properties, List.of(), + List.of(SCHEMA, table), schema, Map.of("kafka.partitions", "5"), false, true); + assertThat(preview.specs).anyMatch(s -> s.contains("kind: KafkaTopic")); + assertThat(preview.specs).anyMatch(s -> s.contains("partitions: 5")); + + // Real create with a different partition count. + HoptimatorDdlUtils.SpecifyResult created = TableService.create(properties, List.of(), + List.of(SCHEMA, table), schema, Map.of("kafka.partitions", "10"), true, false); + assertThat(created.sinkRowType.getFieldNames()).containsExactly("KEY", "VALUE"); + + // Verify it registered by resolving the created topic from the live catalog (mirrors !describe). + RelDataType resolved = CatalogResolver.resolve(List.of(SCHEMA, table)); + assertThat(resolved.getFieldNames()).containsExactly("KEY", "VALUE"); + assertThat(resolved.getField("KEY", false, false).getType().getSqlTypeName()) + .isEqualTo(SqlTypeName.VARCHAR); + assertThat(resolved.getField("VALUE", false, false).getType().getSqlTypeName()) + .isIn(SqlTypeName.BINARY, SqlTypeName.VARBINARY); + + // Drop (cleanup + exercises the delete path). + TableService.delete(properties, List.of(SCHEMA, table)); + } catch (SQLException | RuntimeException e) { + TableService.delete(properties, List.of(SCHEMA, table)); + throw e; + } + } + + @Test + void createExistingTableWithoutUpdateFails() throws SQLException { + String table = "ts-exists-test"; + Schema schema = record("KEY", Schema.Type.STRING, "VALUE", Schema.Type.BYTES); + try { + TableService.create(properties, List.of(), List.of(SCHEMA, table), schema, Map.of(), true, false); + // updateIfExists=false against an existing table must fail. + assertThatThrownBy(() -> + TableService.create(properties, List.of(), List.of(SCHEMA, table), schema, Map.of(), false, false)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("already exists"); + } finally { + TableService.delete(properties, List.of(SCHEMA, table)); + } + } + + @Test + void createInUnknownDatabaseFails() { + Schema schema = record("KEY", Schema.Type.STRING, "VALUE", Schema.Type.BYTES); + assertThatThrownBy(() -> + TableService.create(properties, List.of(), List.of("NOSUCHDB", "t"), schema, Map.of(), true, false)) + .isInstanceOf(SQLException.class); + } + + private static Schema record(String f1, Schema.Type t1, String f2, Schema.Type t2) { + return Schema.createRecord("rec", null, "com.linkedin.hoptimator.test", false, + Arrays.asList(nullable(f1, t1), nullable(f2, t2))); + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } +} diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployer.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployer.java index 624088034..7285f7138 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployer.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployer.java @@ -17,6 +17,7 @@ import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplateList; import com.linkedin.hoptimator.k8s.models.V1alpha1TableTrigger; import com.linkedin.hoptimator.k8s.models.V1alpha1TableTriggerList; +import com.linkedin.hoptimator.util.planner.IdentityQuery; import com.linkedin.hoptimator.util.planner.PipelineRel; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.RelRoot; @@ -354,15 +355,32 @@ public List specify() throws SQLException { */ private Pipeline planPipeline(Source fromSource, Source toSource, String pipelineName) throws Exception { DeploymentContext deploymentContext = context.deploymentContext(); - // Plan SELECT * FROM against the Calcite catalog. - HoptimatorConnection conn = ((CalciteDeploymentContext) deploymentContext).connection(); - Properties props = deploymentContext.properties(); - props.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName); - RelRoot root = HoptimatorDriver.convert(conn, buildSelectSql(fromSource)).root; - final RelNode query = root.rel; - final RelDataType rowType = root.rel.getRowType(); - final ImmutablePairList targetFields = root.fields; - final Map hints = DeploymentService.parseHints(props); + final RelNode query; + final RelDataType rowType; + final ImmutablePairList targetFields; + final Map hints; + if (deploymentContext instanceof CalciteDeploymentContext) { + // SQL path: plan SELECT * FROM against the Calcite catalog. + HoptimatorConnection conn = ((CalciteDeploymentContext) deploymentContext).connection(); + Properties props = deploymentContext.properties(); + props.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName); + RelRoot root = HoptimatorDriver.convert(conn, buildSelectSql(fromSource)).root; + query = root.rel; + rowType = root.rel.getRowType(); + targetFields = root.fields; + hints = DeploymentService.parseHints(props); + } else { + // Connection-free (direct API) path: a logical table's tiers share the row type we created + // both tables from, so the inter-tier pipeline is a pure identity copy — build the scan + // RelNode without a Calcite connection or catalog. + rowType = HoptimatorDriver.rowType(fromSource, deploymentContext); + query = IdentityQuery.scan(fromSource.path(), rowType); + targetFields = IdentityQuery.fields(rowType); + Properties props = new Properties(); + props.putAll(deploymentContext.properties()); + props.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName); + hints = DeploymentService.parseHints(props); + } PipelineRel.Implementor plan = new PipelineRel.Implementor(targetFields, hints); plan.addSource(fromSource.database(), fromSource.path(), rowType, Collections.emptyMap()); plan.setSink(toSource.database(), toSource.path(), rowType, Collections.emptyMap()); diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalOfflineTableServiceIntegrationTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalOfflineTableServiceIntegrationTest.java new file mode 100644 index 000000000..039123b24 --- /dev/null +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalOfflineTableServiceIntegrationTest.java @@ -0,0 +1,104 @@ +package com.linkedin.hoptimator.logical; + +import com.linkedin.hoptimator.jdbc.HoptimatorConnection; +import com.linkedin.hoptimator.jdbc.TableService; +import org.apache.avro.Schema; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Integration tests using the SQL-free {@link TableService} direct API. + * Tests creating a logical table with an offline tier (LOGICAL-OFFLINE = ads-database → ads-catalog-database, + * no Venice) really deploys, and auto-creates a paused {@code TableTrigger} for the offline tier. + * Verified by querying the {@code k8s.table_triggers} metadata table. + */ +@Tag("integration") +public class LogicalOfflineTableServiceIntegrationTest { + + private static final String DB = "LOGICAL-OFFLINE"; + private static final String TABLE = "tsoffline"; + + private HoptimatorConnection connection; + + @BeforeEach + void setUp() throws SQLException { + connection = (HoptimatorConnection) DriverManager.getConnection("jdbc:hoptimator://catalogs=k8s"); + } + + @AfterEach + void tearDown() throws SQLException { + if (connection != null && !connection.isClosed()) { + connection.close(); + } + } + + @Test + void offlineLogicalTableCreatesPausedTrigger() throws SQLException { + try { + // Real create: auto-creates the offline tier physical table + a paused TableTrigger. + create(nullable("ID", Schema.Type.LONG), nullable("NAME", Schema.Type.STRING)); + + // The implicit offline trigger exists, points at the offline physical catalog/schema, paused. + List> triggers = triggersFor(TABLE); + assertThat(triggers).hasSize(1); + assertThat(triggers.get(0)).containsEntry("CATALOG", "ADS_CATALOG"); + assertThat(triggers.get(0)).containsEntry("SCHEMA", "ADS"); + assertThat(triggers.get(0)).containsEntry("PAUSED", "true"); + + // CREATE OR REPLACE (add a column) preserves the trigger. + create(nullable("ID", Schema.Type.LONG), nullable("NAME", Schema.Type.STRING), nullable("EXTRA", Schema.Type.STRING)); + assertThat(triggersFor(TABLE)).hasSize(1); + + // Drop cascades: the trigger is removed. + TableService.delete(connection.connectionProperties(), List.of(DB, TABLE)); + assertThat(triggersFor(TABLE)).isEmpty(); + } catch (SQLException | RuntimeException e) { + TableService.delete(connection.connectionProperties(), List.of(DB, TABLE)); + throw e; + } + } + + private void create(Schema.Field... fields) throws SQLException { + Schema schema = Schema.createRecord(TABLE, null, "com.linkedin.hoptimator.test", false, Arrays.asList(fields)); + TableService.create(connection.connectionProperties(), connection.logHooks(), + List.of(DB, TABLE), schema, Map.of(), true, false); + } + + /** Queries k8s.table_triggers for triggers whose TABLE equals {@code table} (any case). */ + private List> triggersFor(String table) throws SQLException { + List> rows = new ArrayList<>(); + try (Statement st = connection.createStatement(); + ResultSet rs = st.executeQuery( + "select name, catalog, schema, \"TABLE\", paused from \"k8s\".table_triggers")) { + while (rs.next()) { + String t = rs.getString("TABLE"); + if (t != null && t.equalsIgnoreCase(table)) { + rows.add(Map.of( + "NAME", String.valueOf(rs.getString("NAME")), + "CATALOG", String.valueOf(rs.getString("CATALOG")), + "SCHEMA", String.valueOf(rs.getString("SCHEMA")), + "PAUSED", String.valueOf(rs.getBoolean("PAUSED")))); + } + } + } + return rows; + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } +} diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableServiceIntegrationTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableServiceIntegrationTest.java new file mode 100644 index 000000000..c828765e8 --- /dev/null +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableServiceIntegrationTest.java @@ -0,0 +1,168 @@ +package com.linkedin.hoptimator.logical; + +import com.linkedin.hoptimator.jdbc.CatalogResolver; +import com.linkedin.hoptimator.jdbc.HoptimatorConnection; +import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils; +import com.linkedin.hoptimator.jdbc.TableService; +import org.apache.avro.Schema; +import org.apache.calcite.rel.type.RelDataType; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Integration tests using the SQL-free {@link TableService} direct API. + * Asserts the rendered inter-tier pipeline specs; the real-create lifecycle deploys and verifies the + * pipeline CRD/elements (querying the {@code k8s.pipelines} / {@code k8s.pipeline_element_map} + * metadata tables) and the online (Venice) tier's resolved row type. + */ +@Tag("integration") +public class LogicalTableServiceIntegrationTest { + + private static final String DB = "LOGICAL"; + + private HoptimatorConnection connection; + + @BeforeEach + void setUp() throws SQLException { + connection = (HoptimatorConnection) DriverManager.getConnection("jdbc:hoptimator://catalogs=k8s"); + } + + @AfterEach + void tearDown() throws SQLException { + if (connection != null && !connection.isClosed()) { + connection.close(); + } + } + + @Test + void dryRunPlansInterTierPipeline() throws SQLException { + String table = "tslogicaldryrun"; + HoptimatorDdlUtils.SpecifyResult result = TableService.create(connection.connectionProperties(), + connection.logHooks(), List.of(DB, table), + record(table, nullable("KEY", Schema.Type.STRING), nullable("memberId", Schema.Type.LONG), + nullable("pageKey", Schema.Type.STRING)), Map.of(), true, true); + + assertThat(result.sinkRowType.getFieldNames()).containsExactly("KEY", "memberId", "pageKey"); + String specs = String.join("\n", result.specs); + // The nearline (Kafka) and online (Venice) physical tiers plus the identity inter-tier job. + assertThat(specs).contains("kind: FlinkSessionJob"); + assertThat(specs).contains("kind: KafkaTopic"); + assertThat(specs).containsIgnoringCase("insert into"); + } + + @Test + void dryRunPropagatesTierOptions() throws SQLException { + // Options provided to a logical table must reach its physical tiers: kafka.partitions=5 should + // render on the Kafka tier's topic, mirroring KafkaTableServiceIntegrationTest's assertion. + String table = "tslogicalopts"; + HoptimatorDdlUtils.SpecifyResult result = TableService.create(connection.connectionProperties(), + connection.logHooks(), List.of(DB, table), + record(table, nullable("KEY", Schema.Type.STRING), nullable("memberId", Schema.Type.LONG), + nullable("pageKey", Schema.Type.STRING)), + Map.of("kafka.partitions", "5"), true, true); + + String specs = String.join("\n", result.specs); + assertThat(specs).contains("kind: KafkaTopic"); + assertThat(specs).contains("partitions: 5"); + } + + @Test + void createAgainstNonExistentSchemaFails() { + assertThatThrownBy(() -> TableService.create(connection.connectionProperties(), connection.logHooks(), + List.of("LOGICAL-NONEXISTENT", "t"), + record("t", nullable("KEY", Schema.Type.STRING), nullable("id", Schema.Type.LONG)), Map.of(), true, false)) + .isInstanceOf(SQLException.class); + } + + @Test + void onlineLogicalTableLifecycle() throws SQLException { + String table = "tslogicalonline"; + String pipeline = "logical-" + table + "-nearline-to-online"; + try { + // Real create: deploys the Kafka (nearline) + Venice (online) physical tiers and the implicit + // nearline->online identity pipeline. + create(table, nullable("KEY", Schema.Type.STRING), nullable("memberId", Schema.Type.LONG), + nullable("pageKey", Schema.Type.STRING)); + + // The online (Venice) physical tier has a real value schema, so unlike the raw Kafka nearline + // tier it resolves to the full row type. Verify the online store's columns directly. + RelDataType onlineRowType = CatalogResolver.resolve(List.of("VENICE", table)); + assertThat(onlineRowType.getFieldNames()).contains("memberId", "pageKey"); + + // The deployment is also verified structurally via the pipeline CRD and its elements: the + // nearline KafkaTopic physical table plus the identity FlinkSessionJob. + assertThat(pipelineNames()).contains(pipeline); + List elements = pipelineElements(pipeline); + assertThat(elements).anyMatch(e -> e.startsWith("FlinkSessionJob/")); + assertThat(elements).anyMatch(e -> e.startsWith("KafkaTopic/")); + + // CREATE OR REPLACE — add a column; each tier validates backward compatibility. The pipeline + // remains in place and the online tier's schema evolves. + create(table, nullable("KEY", Schema.Type.STRING), nullable("memberId", Schema.Type.LONG), + nullable("pageKey", Schema.Type.STRING), nullable("sessionId", Schema.Type.STRING)); + assertThat(pipelineNames()).contains(pipeline); + assertThat(CatalogResolver.resolve(List.of("VENICE", table)).getFieldNames()).contains("sessionId"); + + // Drop cascades: the implicit pipeline is removed. + TableService.delete(connection.connectionProperties(), List.of(DB, table)); + assertThat(pipelineNames()).doesNotContain(pipeline); + } catch (SQLException | RuntimeException e) { + TableService.delete(connection.connectionProperties(), List.of(DB, table)); + throw e; + } + } + + private void create(String table, Schema.Field... fields) throws SQLException { + TableService.create(connection.connectionProperties(), connection.logHooks(), + List.of(DB, table), record(table, fields), Map.of(), true, false); + } + + private List pipelineNames() throws SQLException { + List names = new ArrayList<>(); + try (Statement st = connection.createStatement(); + ResultSet rs = st.executeQuery("select name from \"k8s\".pipelines")) { + while (rs.next()) { + names.add(rs.getString("NAME")); + } + } + return names; + } + + private List pipelineElements(String pipeline) throws SQLException { + List elements = new ArrayList<>(); + try (Statement st = connection.createStatement(); + ResultSet rs = st.executeQuery( + "select pipeline_name, element_name from \"k8s\".pipeline_element_map")) { + while (rs.next()) { + if (pipeline.equals(rs.getString("PIPELINE_NAME"))) { + elements.add(rs.getString("ELEMENT_NAME")); + } + } + } + return elements; + } + + private static Schema record(String name, Schema.Field... fields) { + return Schema.createRecord(name.replaceAll("\\W", "_"), null, "com.linkedin.hoptimator.test", false, + Arrays.asList(fields)); + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } +} diff --git a/hoptimator-mysql/src/test/java/com/linkedin/hoptimator/mysql/MySqlTableServiceIntegrationTest.java b/hoptimator-mysql/src/test/java/com/linkedin/hoptimator/mysql/MySqlTableServiceIntegrationTest.java new file mode 100644 index 000000000..bc3782999 --- /dev/null +++ b/hoptimator-mysql/src/test/java/com/linkedin/hoptimator/mysql/MySqlTableServiceIntegrationTest.java @@ -0,0 +1,171 @@ +package com.linkedin.hoptimator.mysql; + +import com.linkedin.hoptimator.jdbc.CatalogResolver; +import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils; +import com.linkedin.hoptimator.jdbc.TableService; +import org.apache.avro.Schema; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Integration tests using the SQL-free {@link TableService} direct API. A single lifecycle test groups the + * create/update/drop of one table (verifying registration with {@link CatalogResolver}, mirroring + * {@code !describe}); independent validation/error checks are separate tests. MySQL is a + * catalog-style database ({@code MYSQL.test_database.}). + */ +@Tag("integration") +public class MySqlTableServiceIntegrationTest { + + private static final String CATALOG = "MYSQL"; + private static final String DB = "test_database"; + + @Test + void usersTableLifecycle() throws SQLException { + String table = "ts_users"; + try { + // Create with a KEY_ prefixed primary key. + HoptimatorDdlUtils.SpecifyResult created = create(table, + nullable("KEY_id", Schema.Type.INT), + nullable("name", Schema.Type.STRING), + nullable("email", Schema.Type.STRING)); + assertThat(created.sinkRowType.getFieldNames()).containsExactly("KEY_id", "name", "email"); + + // Verify it registered in MySQL. MySQL maps KEY_ fields to the primary key with the prefix + // stripped, so the physical columns are id/name/email (not KEY_id). + RelDataType resolved = CatalogResolver.resolve(List.of(CATALOG, DB, table)); + assertThat(resolved.getFieldNames()).contains("id", "name", "email"); + assertThat(resolved.getField("id", false, false).getType().getSqlTypeName()) + .isEqualTo(SqlTypeName.INTEGER); + + // Backward-compatible evolution: add a value column (MySQL adds columns, never drops). + HoptimatorDdlUtils.SpecifyResult evolved = create(table, + nullable("KEY_id", Schema.Type.INT), + nullable("name", Schema.Type.STRING), + nullable("email", Schema.Type.STRING), + nullable("age", Schema.Type.INT)); + assertThat(evolved.sinkRowType.getFieldNames()).contains("age"); + assertThat(CatalogResolver.resolve(List.of(CATALOG, DB, table)).getFieldNames()).contains("age"); + + // Drop (cleanup + exercises the delete path). + drop(table); + } catch (SQLException | RuntimeException e) { + drop(table); + throw e; + } + } + + @Test + void ordersCompositeKeyLifecycle() throws SQLException { + String table = "ts_orders"; + try { + HoptimatorDdlUtils.SpecifyResult created = create(table, + nullable("KEY_user_id", Schema.Type.INT), + nullable("KEY_order_id", Schema.Type.INT), + nullable("total", Schema.Type.DOUBLE), + nullable("status", Schema.Type.STRING)); + assertThat(created.sinkRowType.getFieldNames()) + .containsExactly("KEY_user_id", "KEY_order_id", "total", "status"); + + assertThat(CatalogResolver.resolve(List.of(CATALOG, DB, table)).getFieldNames()) + .contains("user_id", "order_id", "total", "status"); + + drop(table); + } catch (SQLException | RuntimeException e) { + drop(table); + throw e; + } + } + + @Test + void createWithoutKeyFieldsFails() { + assertThatThrownBy(() -> create("ts_nokey", + nullable("id", Schema.Type.INT), nullable("data", Schema.Type.STRING))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("No KEY_ fields found in table ts_nokey"); + } + + @Test + void changingKeyFieldsFails() throws SQLException { + String table = "ts_keychange"; + try { + create(table, nullable("KEY_id", Schema.Type.INT), nullable("name", Schema.Type.STRING)); + assertThatThrownBy(() -> create(table, + nullable("KEY_user_id", Schema.Type.INT), nullable("name", Schema.Type.STRING))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Cannot modify KEY fields for table " + table); + } finally { + drop(table); + } + } + + @Test + void changingKeyFieldTypeFails() throws SQLException { + String table = "ts_keytype"; + try { + create(table, nullable("KEY_id", Schema.Type.INT), nullable("name", Schema.Type.STRING)); + assertThatThrownBy(() -> create(table, + nullable("KEY_id", Schema.Type.STRING), nullable("name", Schema.Type.STRING))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Cannot modify KEY field type for table " + table); + } finally { + drop(table); + } + } + + @Test + void createInUnknownCatalogFails() { + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), + List.of("NOSUCHCATALOG", DB, "t"), recordOf("t", nullable("KEY_id", Schema.Type.INT)), + Map.of(), true, false)) + .isInstanceOf(SQLException.class); + } + + @Test + void createWithoutUpdateIfExistsFailsWhenTableExists() throws SQLException { + String table = "ts_exists"; + try { + create(table, nullable("KEY_id", Schema.Type.INT), nullable("name", Schema.Type.STRING)); + // Re-creating the same table with updateIfExists=false must fail rather than silently skip, + // mirroring the SQL path's CREATE (without OR REPLACE) on an existing table. + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), + List.of(CATALOG, DB, table), + recordOf(table, nullable("KEY_id", Schema.Type.INT), nullable("name", Schema.Type.STRING)), + Map.of(), false, false)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("already exists"); + } finally { + drop(table); + } + } + + /** Creates (updateIfExists) a MySQL table at {@code MYSQL.test_database.}. */ + private HoptimatorDdlUtils.SpecifyResult create(String table, Schema.Field... fields) throws SQLException { + return TableService.create(new Properties(), Collections.emptyList(), + List.of(CATALOG, DB, table), recordOf(table, fields), Map.of(), true, false); + } + + private void drop(String table) throws SQLException { + TableService.delete(new Properties(), List.of(CATALOG, DB, table)); + } + + private static Schema recordOf(String table, Schema.Field... fields) { + return Schema.createRecord(table, null, "com.linkedin.hoptimator.test", false, Arrays.asList(fields)); + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } +} diff --git a/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/IdentityQuery.java b/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/IdentityQuery.java new file mode 100644 index 000000000..eda96eba6 --- /dev/null +++ b/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/IdentityQuery.java @@ -0,0 +1,63 @@ +package com.linkedin.hoptimator.util.planner; + +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeField; +import org.apache.calcite.runtime.ImmutablePairList; +import org.apache.calcite.runtime.PairList; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.impl.AbstractSchema; +import org.apache.calcite.schema.impl.AbstractTable; +import org.apache.calcite.tools.FrameworkConfig; +import org.apache.calcite.tools.Frameworks; +import org.apache.calcite.tools.RelBuilder; + +import java.util.List; + + +/** + * Builds an identity query — {@code SELECT * FROM } — as a {@link RelNode}, without a + * Calcite {@code Connection} or catalog. This is for pipelines whose source and sink share a row + * type and therefore need no real query planning, e.g. a logical table's inter-tier copies: because + * both tier tables were created from the same row type, moving data between them is a pure identity + * projection. The resulting {@code RelNode} renders (via {@code RelToSqlConverter}) to the same + * {@code SELECT * FROM catalog.schema.table} the SQL planner would have produced. + */ +public final class IdentityQuery { + + private IdentityQuery() { + } + + /** An identity {@code TableScan} over {@code path} exposing {@code rowType}. */ + public static RelNode scan(List path, RelDataType rowType) { + SchemaPlus root = Frameworks.createRootSchema(false); + SchemaPlus parent = root; + for (String part : path.subList(0, path.size() - 1)) { + parent = parent.add(part, new AbstractSchema()); + } + parent.add(path.get(path.size() - 1), new AbstractTable() { + @Override + public RelDataType getRowType(RelDataTypeFactory factory) { + // Re-home the row type into the builder's type factory to avoid cross-factory issues. + RelDataTypeFactory.Builder builder = factory.builder(); + for (RelDataTypeField field : rowType.getFieldList()) { + builder.add(field.getName(), field.getType()); + } + return builder.build(); + } + }); + FrameworkConfig config = Frameworks.newConfigBuilder().defaultSchema(root).build(); + return RelBuilder.create(config).scan(path).build(); + } + + /** The identity target-field list {@code [(0, f0), (1, f1), ...]} for {@code rowType}. */ + public static ImmutablePairList fields(RelDataType rowType) { + PairList fields = PairList.of(); + int index = 0; + for (RelDataTypeField field : rowType.getFieldList()) { + fields.add(index++, field.getName()); + } + return fields.immutable(); + } +} diff --git a/hoptimator-util/src/test/java/com/linkedin/hoptimator/util/planner/IdentityQueryTest.java b/hoptimator-util/src/test/java/com/linkedin/hoptimator/util/planner/IdentityQueryTest.java new file mode 100644 index 000000000..012d2854b --- /dev/null +++ b/hoptimator-util/src/test/java/com/linkedin/hoptimator/util/planner/IdentityQueryTest.java @@ -0,0 +1,64 @@ +package com.linkedin.hoptimator.util.planner; + +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeSystem; +import org.apache.calcite.runtime.ImmutablePairList; +import org.apache.calcite.sql.type.SqlTypeFactoryImpl; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + + +class IdentityQueryTest { + + private static RelDataType rowType() { + RelDataTypeFactory factory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT); + return factory.builder() + .add("ID", factory.createSqlType(SqlTypeName.INTEGER)) + .add("NAME", factory.createSqlType(SqlTypeName.VARCHAR)) + .build(); + } + + @Test + void scanBuildsIdentityScanExposingRowType() { + List path = Arrays.asList("CATALOG", "SCHEMA", "TABLE"); + + RelNode scan = IdentityQuery.scan(path, rowType()); + + assertThat(scan).isNotNull(); + // The scan re-homes the carried row type; field names/count must round-trip. + assertThat(scan.getRowType().getFieldNames()).containsExactly("ID", "NAME"); + assertThat(scan.getTable().getQualifiedName()).containsExactly("CATALOG", "SCHEMA", "TABLE"); + } + + @Test + void scanWorksForSingleSchemaPath() { + RelNode scan = IdentityQuery.scan(Arrays.asList("SCHEMA", "TABLE"), rowType()); + + assertThat(scan.getTable().getQualifiedName()).containsExactly("SCHEMA", "TABLE"); + assertThat(scan.getRowType().getFieldNames()).containsExactly("ID", "NAME"); + } + + @Test + void fieldsReturnsIndexedTargetList() { + ImmutablePairList fields = IdentityQuery.fields(rowType()); + + assertThat(fields).hasSize(2); + assertThat(fields.leftList()).containsExactly(0, 1); + assertThat(fields.rightList()).containsExactly("ID", "NAME"); + } + + @Test + void fieldsOfEmptyRowTypeIsEmpty() { + RelDataTypeFactory factory = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT); + RelDataType empty = factory.builder().build(); + + assertThat(IdentityQuery.fields(empty)).isEmpty(); + } +} diff --git a/hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDeployer.java b/hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDeployer.java index e258dd64b..d1c5b3016 100644 --- a/hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDeployer.java +++ b/hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDeployer.java @@ -199,6 +199,12 @@ public void restore() { protected Pair getKeyPayloadSchema() throws SQLException { Map keyOptions = resolveKeyOptions(); + // On the direct API path the caller supplied the exact Avro schema; split it losslessly instead + // of re-synthesizing from the row type (which would drop namespaces, nested record names, ...). + Schema provided = HoptimatorDriver.providedAvroSchema(context); + if (provided != null) { + return AvroConverter.avroKeyPayloadSchema(source.table() + "_Key", provided, keyOptions); + } return AvroConverter.avroKeyPayloadSchema("com.linkedin.hoptimator", source.table() + "_Key", source.table() + "_Value", diff --git a/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceDeployerTest.java b/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceDeployerTest.java index e8eb5e1cf..a9be7e242 100644 --- a/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceDeployerTest.java +++ b/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceDeployerTest.java @@ -3,6 +3,7 @@ import com.linkedin.hoptimator.Source; import com.linkedin.hoptimator.Validator; import com.linkedin.hoptimator.jdbc.CalciteDeploymentContext; +import com.linkedin.hoptimator.jdbc.DirectDeploymentContext; import com.linkedin.hoptimator.jdbc.HoptimatorConnection; import com.linkedin.venice.client.schema.StoreSchemaFetcher; import com.linkedin.venice.controllerapi.ControllerClient; @@ -36,6 +37,7 @@ import java.util.Map; import java.util.Properties; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -323,6 +325,36 @@ public void testExistsReturnsFalseWhenStoreAbsent() throws Exception { assertFalse(createDeployer(source).exists()); } + @Test + public void testGetKeyPayloadSchemaPrefersProvidedAvroLosslessly() throws Exception { + Source source = new Source("venice", List.of("VENICE", TEST_STORE), Collections.emptyMap()); + // The direct path carries the caller's Avro. getKeyPayloadSchema must use it verbatim rather + // than re-synthesizing from the row type, so a nested value record's namespace survives. + Schema provided = new Schema.Parser().parse("{" + + "\"type\":\"record\",\"name\":\"StoreValue\",\"namespace\":\"com.example.venice\",\"fields\":[" + + "{\"name\":\"widget\",\"type\":{\"type\":\"record\",\"name\":\"Widget\"," + + "\"namespace\":\"com.example.custom\",\"fields\":[{\"name\":\"w\",\"type\":\"string\"}]}}" + + "]}"); + VeniceDeployer deployer = new VeniceDeployer( + source, properties, new DirectDeploymentContext(properties, null, provided)) { + @Override + protected Map resolveKeyOptions() { + // Unit-test seam: avoid resolving options through the live ConnectionService (which would + // reach out to K8s). The provided-Avro path under test does not depend on key options. + return Collections.emptyMap(); + } + }; + + Pair keyPayload = deployer.getKeyPayloadSchema(); + + Schema payload = keyPayload.right; + assertNotNull(payload, "payload schema"); + assertEquals("com.example.venice", payload.getNamespace()); + Schema nested = payload.getField("widget").schema(); + assertEquals("com.example.custom", nested.getNamespace(), "nested namespace preserved losslessly"); + assertEquals("Widget", nested.getName()); + } + @Test public void testGetKeyPayloadSchemaProducesPayloadFromRowType() throws Exception { Source source = new Source("venice", List.of("VENICE", TEST_STORE), Collections.emptyMap()); diff --git a/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceTableServiceIntegrationTest.java b/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceTableServiceIntegrationTest.java new file mode 100644 index 000000000..c085af137 --- /dev/null +++ b/hoptimator-venice/src/test/java/com/linkedin/hoptimator/venice/VeniceTableServiceIntegrationTest.java @@ -0,0 +1,249 @@ +package com.linkedin.hoptimator.venice; + +import com.linkedin.hoptimator.Source; +import com.linkedin.hoptimator.jdbc.DatabaseConfigResolvers; +import com.linkedin.hoptimator.jdbc.DirectDeploymentContext; +import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils; +import com.linkedin.hoptimator.jdbc.TableService; +import com.linkedin.venice.client.schema.StoreSchemaFetcher; +import org.apache.avro.Schema; +import org.apache.calcite.jdbc.CalciteConnection; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.Table; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Integration tests using the SQL-free {@link TableService} direct API. + * Exercises Venice CREATE TABLE key/value schema derivation, backward-compatible evolution and validation failures. + */ +@Tag("integration") +public class VeniceTableServiceIntegrationTest { + + private static final String SCHEMA = "VENICE"; + + @Test + void storeLifecycle() throws SQLException { + String store = "directapi-store"; + try { + // Create with a record (KEY_ prefixed) key plus value fields. + HoptimatorDdlUtils.SpecifyResult created = create(store, + nullable("KEY_id", Schema.Type.INT), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING)); + assertThat(created.sinkRowType.getFieldNames()).containsExactly("KEY_id", "i", "s"); + + // Verify it registered in Venice. + RelDataType resolved = resolveVenice(store); + assertThat(resolved.getFieldNames()).contains("i", "s"); + + // Backward-compatible evolution: add a nullable value field. + HoptimatorDdlUtils.SpecifyResult evolved = create(store, + nullable("KEY_id", Schema.Type.INT), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING), + nullable("new_field", Schema.Type.DOUBLE)); + assertThat(evolved.sinkRowType.getFieldNames()).contains("new_field"); + assertThat(resolveVenice(store).getFieldNames()).contains("new_field"); + + // Invalid updates on the same store must fail. + assertThatThrownBy(() -> create(store, + nullable("KEY_user_id", Schema.Type.INT), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING), + nullable("new_field", Schema.Type.DOUBLE))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Key schema evolution is not supported in Venice"); + + assertThatThrownBy(() -> create(store, + nullable("KEY_id", Schema.Type.STRING), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING), + nullable("new_field", Schema.Type.DOUBLE))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Key schema evolution is not supported in Venice"); + + assertThatThrownBy(() -> create(store, + nullable("KEY_id", Schema.Type.INT), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING), + required("new_field", Schema.Type.DOUBLE))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Value schema is not backward compatible"); + } finally { + drop(store); + } + } + + @Test + void compositeKeyStoreLifecycle() throws SQLException { + String store = "directapi-store-composite"; + try { + HoptimatorDdlUtils.SpecifyResult created = create(store, + nullable("KEY_user_id", Schema.Type.INT), + nullable("KEY_order_id", Schema.Type.INT), + nullable("total", Schema.Type.DOUBLE), + nullable("status", Schema.Type.STRING)); + assertThat(created.sinkRowType.getFieldNames()) + .containsExactly("KEY_user_id", "KEY_order_id", "total", "status"); + assertThat(resolveVenice(store).getFieldNames()) + .contains("total", "status"); + } finally { + drop(store); + } + } + + @Test + void primitiveKeyStoreLifecycle() throws SQLException { + String store = "directapi-store-primitive"; + try { + HoptimatorDdlUtils.SpecifyResult created = create(store, + nullable("KEY", Schema.Type.INT), + nullable("i", Schema.Type.INT), + nullable("s", Schema.Type.STRING)); + assertThat(created.sinkRowType.getFieldNames()).containsExactly("KEY", "i", "s"); + assertThat(resolveVenice(store).getFieldNames()).contains("i", "s"); + } finally { + drop(store); + } + } + + @Test + void createWithoutKeyFieldsFails() { + assertThatThrownBy(() -> create("directapi-nokey", + nullable("i", Schema.Type.INT), nullable("s", Schema.Type.STRING))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Failed to generate key schema for Venice store directapi-nokey"); + } + + @Test + void createWithoutValueFieldsFails() { + assertThatThrownBy(() -> create("directapi-novalue", nullable("KEY", Schema.Type.INT))) + .isInstanceOf(SQLException.class) + .hasMessageContaining("Failed to generate value schema for Venice store directapi-novalue"); + } + + @Test + void createInUnknownDatabaseFails() { + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), + List.of("NOSUCHDB", "t"), recordOf("t", nullable("KEY", Schema.Type.INT), nullable("i", Schema.Type.INT)), + Map.of(), true, false)) + .isInstanceOf(SQLException.class); + } + + @Test + void providedValueSchemaNamespaceIsPreservedInVenice() throws Exception { + // End-to-end proof that the direct API deploys the caller's Avro verbatim (no lossy + // Avro->RelDataType->Avro round-trip): a value field whose type is a nested record in its own + // namespace must come back from Venice with that namespace intact. + String store = "directapi-nsvalue"; + try { + Schema.Field widget = new Schema.Field("widget", + Schema.createRecord("Widget", null, "com.example.custom", false, + List.of(new Schema.Field("w", Schema.create(Schema.Type.STRING), null, null))), + null, null); + TableService.create(new Properties(), Collections.emptyList(), List.of(SCHEMA, store), + recordOf(sanitize(store), nullable("KEY_id", Schema.Type.INT), widget), + Map.of(), true, false); + + // Read the value schema back from real Venice and assert the nested namespace survived. + Properties veniceProps = DatabaseConfigResolvers.forProperties(new Properties()) + .databaseProperties(null, SCHEMA, "jdbc:venice://"); + Source source = new Source(store, List.of(SCHEMA, store), Map.of()); + VeniceDeployer reader = new VeniceDeployer(source, veniceProps, + new DirectDeploymentContext(veniceProps, null, null)); + try (StoreSchemaFetcher fetcher = reader.createStoreSchemaFetcher(store)) { + Schema registeredValue = fetcher.getLatestValueSchema(); + Schema nested = registeredValue.getField("widget").schema(); + assertThat(nested.getName()).isEqualTo("Widget"); + assertThat(nested.getNamespace()).isEqualTo("com.example.custom"); + } + } finally { + drop(store); + } + } + + @Test + void createWithoutUpdateIfExistsFailsWhenStoreExists() throws SQLException { + String store = "directapi-exists"; + try { + create(store, nullable("KEY", Schema.Type.INT), nullable("i", Schema.Type.INT)); + // Re-creating the same store with updateIfExists=false must fail rather than silently skip. + assertThatThrownBy(() -> TableService.create(new Properties(), Collections.emptyList(), + List.of(SCHEMA, store), + recordOf(sanitize(store), nullable("KEY", Schema.Type.INT), nullable("i", Schema.Type.INT)), + Map.of(), false, false)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("already exists"); + } finally { + drop(store); + } + } + + private HoptimatorDdlUtils.SpecifyResult create(String store, Schema.Field... fields) throws SQLException { + return TableService.create(new Properties(), Collections.emptyList(), + List.of(SCHEMA, store), recordOf(sanitize(store), fields), Map.of(), true, false); + } + + private void drop(String store) throws SQLException { + TableService.delete(new Properties(), List.of(SCHEMA, store)); + } + + /** + * Resolves a Venice store's row type (key fields, {@code KEY_}-prefixed or {@code KEY} for a + * primitive key, followed by the value fields), failing if the store is not present. + * + *

Uses a direct {@code jdbc:venice://} connection rather than the shared {@code catalogs=k8s} + * catalog on purpose. The catalog resolves a table by enumerating the whole schema, which for + * Venice means the router's bulk {@code /stores} list — a periodically-refreshed cache that lags + * the controller, so a just-created store can be momentarily absent from it (and, because that + * refresh window can outlast a whole test run, absent for every store created that run — the + * "all three or none" flake). A direct connection instead resolves the single store via the + * router's targeted {@code discover_cluster} endpoint, which reflects the create immediately, so + * no polling is needed. The store deployers run synchronously, so the store exists the instant + * {@code create} returns. + */ + private RelDataType resolveVenice(String store) throws SQLException { + Properties veniceProps = DatabaseConfigResolvers.forProperties(new Properties()) + .databaseProperties(null, SCHEMA, "jdbc:venice://"); + try (Connection conn = DriverManager.getConnection("jdbc:venice://", veniceProps)) { + CalciteConnection calciteConnection = conn.unwrap(CalciteConnection.class); + SchemaPlus veniceSchema = calciteConnection.getRootSchema().subSchemas().get(SCHEMA); + Table table = veniceSchema == null ? null : veniceSchema.tables().get(store); + if (table == null) { + throw new SQLException("Venice store not resolvable after create: " + store); + } + return table.getRowType(calciteConnection.getTypeFactory()); + } + } + + private static Schema recordOf(String name, Schema.Field... fields) { + return Schema.createRecord(name, null, "com.linkedin.hoptimator.test", false, Arrays.asList(fields)); + } + + private static Schema.Field nullable(String name, Schema.Type type) { + Schema union = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(type)); + return new Schema.Field(name, union, null, Schema.Field.NULL_DEFAULT_VALUE); + } + + private static Schema.Field required(String name, Schema.Type type) { + return new Schema.Field(name, Schema.create(type), null, null); + } + + private static String sanitize(String name) { + return name.replaceAll("\\W", "_"); + } +}