diff --git a/README.md b/README.md index 7ebaa988..19098353 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,7 @@ but `hoptimator-api` is the actual extension point. - **Multi-hop, declarative.** You don't write Flink jobs and you don't request topics. The planner figures out the topology from a query. - **Kubernetes out of the box, not as a hard requirement.** The bundled - deployers target Kubernetes, so pipelines show up as first-class CRDs and + deployers target Kubernetes, so pipelines show up as first-class custom resources and `kubectl get pipelines` Just Works. The `Deployer` interface is the actual extension point — anything that knows how to materialize a spec can take the place of the defaults. diff --git a/docs/extending/data-sources.md b/docs/extending/data-sources.md index ded92df5..a8687af1 100644 --- a/docs/extending/data-sources.md +++ b/docs/extending/data-sources.md @@ -4,7 +4,7 @@ Adding a new external system to Hoptimator usually means three things: 1. A **JDBC adapter** that exposes the system's tables and schemas through a JDBC connection — Hoptimator reads metadata through this. -2. A **`Database` CRD** that registers the adapter (a JDBC URL plus a +2. A **`Database` custom resource** that registers the adapter (a JDBC URL plus a schema name) so the catalog includes it. 3. A **`TableTemplate`** (and possibly a **`JobTemplate`**) that tells Hoptimator how to deploy resources for the system — Kafka topics, Venice @@ -53,7 +53,7 @@ com.example.hoptimator.mysystem.MySystemDriver ## Registering with the catalog -Once your driver is on the classpath, a `Database` CRD makes it visible to +Once your driver is on the classpath, a `Database` custom resource makes it visible to Hoptimator: ```yaml @@ -72,7 +72,7 @@ your driver expects. See the [Database CRD reference](../kubernetes/crd-reference.md#database) for all fields. -After applying the CRD, Hoptimator's catalog picks it up on the next +After applying the custom resource, Hoptimator's catalog picks it up on the next connection — `!tables` in the SQL CLI should show your system's tables. ## Telling Hoptimator how to deploy resources @@ -107,13 +107,13 @@ spec: ``` No Java. The bundled Kafka deployment uses this pattern — Hoptimator emits -a `KafkaTopic` CRD; Strimzi creates the topic. See +a `KafkaTopic` custom resource; Strimzi creates the topic. See [Templates and configuration](../kubernetes/templates.md) for the placeholder syntax and matching rules. ### Path B: imperative, via a custom Deployer -If your system needs an admin API call rather than a YAML CRD apply (e.g. +If your system needs an admin API call rather than a YAML custom resource apply (e.g. calling a Venice controller's REST endpoint), you'll need a custom `Deployer`. This is the path the bundled `hoptimator-venice` and `hoptimator-kafka` modules take in addition to (or instead of) templates. diff --git a/docs/extending/deployers.md b/docs/extending/deployers.md index f3c394db..8a6a7d58 100644 --- a/docs/extending/deployers.md +++ b/docs/extending/deployers.md @@ -9,7 +9,7 @@ with a spec" can plug in. You'll need a deployer when: - You're integrating a system whose resources are created via an admin API - (REST, gRPC, command-line) rather than a CRD apply. Templates aren't + (REST, gRPC, command-line) rather than a custom resource apply. Templates aren't enough — you need imperative code. - You want pipelines to deploy to something other than Kubernetes — a Nomad cluster, an external service registry, your own internal control @@ -116,7 +116,7 @@ The bundled Kafka path is a good shape to copy: - [`KafkaDeployerProvider`](https://github.com/linkedin/Hoptimator/blob/main/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java) — type-checks the `Deployable`, resolves the `Database`'s connection config via `context.databaseProperties(catalog, schema, "jdbc:kafka://")` - (the JDBC URL the `Database` CRD points at), and constructs the deployer. + (the JDBC URL the `Database` custom resource points at), and constructs the deployer. - [`KafkaDeployer`](https://github.com/linkedin/Hoptimator/blob/main/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployer.java) — `create()` calls Kafka's AdminClient API to create the topic; `restore()` walks back and deletes any topic the current operation diff --git a/docs/extending/index.md b/docs/extending/index.md index ee82fc8d..d290f642 100644 --- a/docs/extending/index.md +++ b/docs/extending/index.md @@ -1,7 +1,7 @@ # Extending Hoptimator Hoptimator's behavior is driven by Java SPI plug-ins (`ServiceLoader`-based) -and by the `TableTemplate` / `JobTemplate` CRDs. Most extensions don't need +and by the `TableTemplate` / `JobTemplate` custom resources. Most extensions don't need both — pick the layer that matches what you're doing. ## Pick the right surface @@ -47,7 +47,7 @@ For surfaces that produce multiple values for the same input — `Validator`, ### "I just want to add my system to the catalog" -The lowest-friction path is **a JDBC driver + a `Database` CRD**. Hoptimator +The lowest-friction path is **a JDBC driver + a `Database` custom resource**. Hoptimator treats anything that responds to a JDBC URL as a potential catalog source. You point a `Database` at it, and Hoptimator pulls schemas and tables from that connection. See [Data sources → Adapter](data-sources.md#the-jdbc-adapter). @@ -55,7 +55,7 @@ that connection. See [Data sources → Adapter](data-sources.md#the-jdbc-adapter ### "I need Hoptimator to actually deploy my system's resources" After the adapter, ship a `TableTemplate` (or `JobTemplate`) that emits the -YAML for your storage system's CRD or operator. Templates are a CRD, so +YAML for your storage system's custom resource or operator. Templates are a custom resource, so this is YAML-only — no Java needed. See [Templates and configuration](../kubernetes/templates.md). diff --git a/docs/extending/validators.md b/docs/extending/validators.md index a0b28ca2..2268dc4d 100644 --- a/docs/extending/validators.md +++ b/docs/extending/validators.md @@ -1,6 +1,6 @@ # Validators -A `Validator` inspects a SQL statement, a CRD, or a planned pipeline element +A `Validator` inspects a SQL statement, a custom resource, or a planned pipeline element *before* anything is deployed and rejects it if it doesn't meet your constraints. Validators are the right place for environment-specific policy: naming conventions, schema compatibility checks, ACL enforcement, anything @@ -20,7 +20,7 @@ DDL path: If any validator emits an `Issues.error(...)` at any of those points, the whole operation aborts and the error message surfaces to the user (or to -the operator's status field for CRD-driven changes). +the operator's status field for custom-resource-driven changes). ## The interfaces diff --git a/docs/getting-started/architecture.md b/docs/getting-started/architecture.md index 6b37fb4f..b1afa1fc 100644 --- a/docs/getting-started/architecture.md +++ b/docs/getting-started/architecture.md @@ -74,7 +74,7 @@ deploy the operator standalone and feed it Subscriptions from CI. ``` The same flow applies whether you start from SQL (CLI, JDBC, MCP) or from a -`Subscription` CRD applied with `kubectl apply -f`. +`Subscription` custom resource applied with `kubectl apply -f`. ## Step 1 — Parse and resolve @@ -164,7 +164,7 @@ supplies a Calcite-backed `DeploymentContext` (row type read from the catalog); the direct path supplies a `DirectDeploymentContext` that carries the caller's Avro schema (deriving the row type on demand) and resolves `Database` config registry-natively via a `DatabaseConfigResolver` -(the K8s implementation reads `Database` CRDs directly — see +(the K8s implementation reads `Database` custom resources directly — see `DatabaseConfigResolvers`). The direct path opens no connection and touches no Calcite catalog; only the SQL engine's read/plan path still uses the JDBC driver layer. @@ -205,7 +205,7 @@ contributions should not target them. the runtime's operator already understands (e.g. a Beam `FlinkSessionJob`, a Spark Operator `SparkApplication`). Hoptimator generates the spec; the target operator runs the job. No `Engine` registration is required for - this path — the `Engine` CRD is only needed if Hoptimator should submit + this path — the `Engine` custom resource is only needed if Hoptimator should submit *queries* (not pipelines) directly to the runtime. - **A new deployment target**: implement `Deployer` and register it via `DeployerProvider`. Kubernetes is the default but not a hard requirement; diff --git a/docs/getting-started/concepts.md b/docs/getting-started/concepts.md index 4b2e6824..3edb3337 100644 --- a/docs/getting-started/concepts.md +++ b/docs/getting-started/concepts.md @@ -12,23 +12,23 @@ the documentation will read more naturally. ## At a glance -| Concept | What it is | -| ------------------- | ------------------------------------------------------------------------------------------------ | -| **Database** | A connection to an external system that exposes tables (Kafka, Venice, MySQL, etc.). | -| **Catalog** | The unified namespace that lets a single SQL statement reference tables across many databases. | -| **View** | A named SQL query, evaluated lazily. | -| **Materialized view** | A view backed by a running data pipeline that continuously writes results to a sink. | -| **Pipeline** | The set of sources, sink, and job that together implement a materialized view. | -| **Engine** | A runtime Hoptimator can submit *queries* to (e.g. a Flink SQL gateway). Optional. Pipeline materialization does *not* require one. | -| **Connector** | Configuration that tells a runtime how to read from or write to a database. Used by the planner and embedded in template output. | -| **Deployer** | The component that turns a planned pipeline element into real infrastructure. | -| **Validator** | Pre-deploy check that rejects SQL, CRDs, or planned pipelines that violate environment policy. | -| **TableTemplate** | Declarative recipe for materializing a source/sink in a particular database. | -| **JobTemplate** | Declarative recipe for materializing a job on a particular engine. | -| **TableTrigger** | Fires a Kubernetes job when an upstream table changes (or on a schedule). | -| **LogicalTable** | An abstraction model: one named entity that physically lives in many backends (nearline / online / offline). Auto-syncs and auto-backfills its tiers. | -| **Subscription** | YAML-native way to declare a materialized view; equivalent to `CREATE MATERIALIZED VIEW ... AS`. | -| **Hint** | Key/value passed at runtime that templates and connectors can pick up. | +| Concept | What it is | +|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------| +| **Database** | A connection to an external system that exposes tables (Kafka, Venice, MySQL, etc.). | +| **Catalog** | The unified namespace that lets a single SQL statement reference tables across many databases. | +| **View** | A named SQL query, evaluated lazily. | +| **Materialized view** | A view backed by a running data pipeline that continuously writes results to a sink. | +| **Pipeline** | The set of sources, sink, and job that together implement a materialized view. | +| **Engine** | A runtime Hoptimator can submit *queries* to (e.g. a Flink SQL gateway). Optional. Pipeline materialization does *not* require one. | +| **Connector** | Configuration that tells a runtime how to read from or write to a database. Used by the planner and embedded in template output. | +| **Deployer** | The component that turns a planned pipeline element into real infrastructure. | +| **Validator** | Pre-deploy check that rejects SQL, custom resources, or planned pipelines that violate environment policy. | +| **TableTemplate** | Declarative recipe for materializing a source/sink in a particular database. | +| **JobTemplate** | Declarative recipe for materializing a job on a particular engine. | +| **TableTrigger** | Fires a Kubernetes job when an upstream table changes (or on a schedule). | +| **LogicalTable** | An abstraction model: one named entity that physically lives in many backends (nearline / online / offline). Auto-syncs and auto-backfills its tiers. | +| **Subscription** | YAML-native way to declare a materialized view; equivalent to `CREATE MATERIALIZED VIEW ... AS`. | +| **Hint** | Key/value passed at runtime that templates and connectors can pick up. | ## Databases, schemas, and tables @@ -108,7 +108,7 @@ Operator — picks it up and runs the job. Hoptimator is not in the data path. ## Engines (optional) -An **Engine** CRD registers a runtime Hoptimator can submit **queries** to — +An **Engine** custom resource registers a runtime Hoptimator can submit **queries** to — typically a Flink SQL gateway behind a JDBC URL. This is the path used when Hoptimator needs to *execute* SQL itself, e.g. for interactive `SELECT` against tables that aren't in-process. @@ -136,7 +136,7 @@ See [Extending Hoptimator](../extending/index.md) when those docs land. ## Validators -A **Validator** inspects a SQL statement, a CRD, or a planned pipeline +A **Validator** inspects a SQL statement, a custom resource, or a planned pipeline element *before* it deploys and rejects it if it doesn't meet your constraints. Where `Deployer` is "make this real," `Validator` is "check this is allowed." @@ -256,7 +256,7 @@ infrastructure for each binding: the normal Deployer SPI to create whatever the storage system needs (a Kafka topic, a Venice store, an HDFS dataset). - **Implicit inter-tier pipelines.** Hoptimator auto-deploys - `nearline → online` and `nearline → offline` Pipeline CRDs to keep the + `nearline → online` and `nearline → offline` Pipeline custom resources to keep the tiers consistent. You don't write the Kafka-to-Venice job; it appears because the LogicalTable says it should. - **Auto-backfill triggers.** When an offline tier is present, a diff --git a/docs/index.md b/docs/index.md index c19b9739..d5ffca3d 100644 --- a/docs/index.md +++ b/docs/index.md @@ -58,7 +58,7 @@ See the **[Kubernetes guide](kubernetes/index.md)**: See **[Extending Hoptimator](extending/index.md)**: - [Adding a new data source](extending/data-sources.md) — JDBC adapter, - `Database` CRD, `TableTemplate` authoring. + `Database` custom resource, `TableTemplate` authoring. - [Deployers](extending/deployers.md) — implementing `Deployer` for a new deployment target, `DeployerProvider` registration, lifecycle. - [Validators](extending/validators.md) — pre-deploy policy enforcement diff --git a/docs/kubernetes/crd-reference.md b/docs/kubernetes/crd-reference.md index 0d7342ba..df92c44f 100644 --- a/docs/kubernetes/crd-reference.md +++ b/docs/kubernetes/crd-reference.md @@ -360,16 +360,16 @@ spec: ### Spec fields -| Field | Type | Description | -| ----------- | ------ | -------------------------------------------------------------------------------------- | -| `tableName` | string | Original table name as declared in `CREATE TABLE` (e.g. `audience`). | -| `tiers` | object | Map of tier name (`nearline`, `online`, `offline`) to a tier binding. | +| Field | Type | Description | +|-------------|--------|-----------------------------------------------------------------------| +| `tableName` | string | Original table name as declared in `CREATE TABLE` (e.g. `audience`). | +| `tiers` | object | Map of tier name (`nearline`, `online`, `offline`) to a tier binding. | Each tier binding has one field: -| Field | Type | Required | Description | -| ---------- | ------ | :------: | ------------------------------------------------- | -| `database` | string | yes | Name of the `Database` CRD backing this tier. | +| Field | Type | Required | Description | +|------------|--------|:--------:|-----------------------------------------------------------| +| `database` | string | yes | Name of the `Database` custom resource backing this tier. | The `LogicalTableDeployer` runs at create time to deploy physical tier resources, the implicit inter-tier sync pipelines, and the offline-tier diff --git a/docs/kubernetes/index.md b/docs/kubernetes/index.md index f105de0e..7b3ddd3c 100644 --- a/docs/kubernetes/index.md +++ b/docs/kubernetes/index.md @@ -29,4 +29,4 @@ operator reconciles them. - [Architecture](../getting-started/architecture.md) — what the operator is doing in the bigger picture. - [DDL reference](../user-guide/ddl-reference.md) — SQL DDL that has YAML - CRD equivalents. + custom resource equivalents. diff --git a/docs/kubernetes/operator.md b/docs/kubernetes/operator.md index 6fa55fa9..e4fa45ca 100644 --- a/docs/kubernetes/operator.md +++ b/docs/kubernetes/operator.md @@ -1,7 +1,7 @@ # The operator `hoptimator-operator` is the long-running Kubernetes controller that -reconciles Hoptimator's CRDs. It uses the same Deployer machinery as the +reconciles Hoptimator's custom resources. It uses the same Deployer machinery as the SQL path — when a `Subscription`, `Pipeline`, `View`, or `TableTrigger` changes, it asks the deployers to bring the cluster state in line with the spec. @@ -47,7 +47,7 @@ command: ## Namespace scoping -By default, the operator watches **all namespaces** for the CRDs it owns. +By default, the operator watches **all namespaces** for the custom resources it owns. To restrict it to one namespace, pass `--watch `: ```yaml @@ -143,8 +143,8 @@ Logs are the primary debugging surface today. ## When *not* to run the operator The operator is only required when you want continuous reconciliation — -typically when applying CRDs via `kubectl` rather than driving everything +typically when applying custom resources via `kubectl` rather than driving everything through the JDBC path. If your workflow is "developer runs `./hoptimator` -to create materialized views" and nothing else applies CRDs, you don't +to create materialized views" and nothing else applies custom resources, you don't strictly need the operator running. The CLI deploys synchronously and waits for success. diff --git a/docs/kubernetes/templates.md b/docs/kubernetes/templates.md index b99d9d52..d8f3d3c9 100644 --- a/docs/kubernetes/templates.md +++ b/docs/kubernetes/templates.md @@ -401,18 +401,18 @@ can be set on the JDBC URL (`jdbc:hoptimator://k8s.namespace=my-team`), passed in a `Properties` object to `DriverManager.getConnection`, or, in the operator deployment, baked into the Calcite model file. -| Property | Default | Description | -| --------------------------------- | -------------------------------- | ---------------------------------------------------------------------------------------- | -| `k8s.namespace` | the active namespace | Namespace Hoptimator reads CRDs from and writes deployed resources to. | -| `k8s.watch.namespace` | same as `k8s.namespace` | Namespace the operator watches for reconciliation. Empty string means all namespaces. | -| `k8s.kubeconfig` | `$KUBECONFIG` / `~/.kube/config` | Path to a kubeconfig file. | -| `k8s.server` | from kubeconfig | API server URL. Required with `k8s.token` or `k8s.user`+`k8s.password`. | -| `k8s.user` / `k8s.password` | *(none)* | Basic-auth credentials. Requires `k8s.server`. | -| `k8s.token` | *(none)* | Bearer token. Requires `k8s.server`. | -| `k8s.impersonate.user` | *(none)* | Impersonate this user when calling the API. | -| `k8s.impersonate.group` | *(none)* | Single impersonation group. | -| `k8s.impersonate.groups` | *(none)* | Comma-separated impersonation groups. | -| `k8s.ssl.truststore.location` | *(none)* | Path to a PEM/JKS truststore for the API server certificate. | +| Property | Default | Description | +|-------------------------------|----------------------------------|---------------------------------------------------------------------------------------| +| `k8s.namespace` | the active namespace | Namespace Hoptimator reads custom resources from and writes deployed resources to. | +| `k8s.watch.namespace` | same as `k8s.namespace` | Namespace the operator watches for reconciliation. Empty string means all namespaces. | +| `k8s.kubeconfig` | `$KUBECONFIG` / `~/.kube/config` | Path to a kubeconfig file. | +| `k8s.server` | from kubeconfig | API server URL. Required with `k8s.token` or `k8s.user`+`k8s.password`. | +| `k8s.user` / `k8s.password` | *(none)* | Basic-auth credentials. Requires `k8s.server`. | +| `k8s.token` | *(none)* | Bearer token. Requires `k8s.server`. | +| `k8s.impersonate.user` | *(none)* | Impersonate this user when calling the API. | +| `k8s.impersonate.group` | *(none)* | Single impersonation group. | +| `k8s.impersonate.groups` | *(none)* | Comma-separated impersonation groups. | +| `k8s.ssl.truststore.location` | *(none)* | Path to a PEM/JKS truststore for the API server certificate. | If none of the above are set, the driver behaves like `kubectl` would: it reads `~/.kube/config` and uses the active context. diff --git a/docs/user-guide/ddl-reference.md b/docs/user-guide/ddl-reference.md index 871a3daa..8ab4e8c3 100644 --- a/docs/user-guide/ddl-reference.md +++ b/docs/user-guide/ddl-reference.md @@ -6,7 +6,7 @@ ANSI SQL — see the This page documents the DDL Hoptimator adds on top. > All DDL listed here also has a YAML equivalent: a `View`, `Pipeline`, -> `TableTrigger`, etc. CRD. Use whichever is more ergonomic for your workflow. +> `TableTrigger`, etc. custom resource. Use whichever is more ergonomic for your workflow. > Test cases for the DDL parser and executor live as > [Quidem](https://github.com/julianhyde/quidem) `.id` scripts under each @@ -157,7 +157,7 @@ CREATE [OR REPLACE] TRIGGER [IF NOT EXISTS] [WITH ('' '', ...)] ``` -Equivalent to a `TableTrigger` CRD: runs the embedded YAML (typically a Job +Equivalent to a `TableTrigger` custom resource: runs the embedded YAML (typically a Job or CronJob) when the named table changes or on a cron schedule. The job spec is arbitrary, so triggers are how you wire up backfills, rETL refreshes, downstream notifications, and operational hooks without embedding that diff --git a/docs/user-guide/hints.md b/docs/user-guide/hints.md index d8b17fae..a4d7e65a 100644 --- a/docs/user-guide/hints.md +++ b/docs/user-guide/hints.md @@ -15,7 +15,7 @@ than you expected, hints that no longer apply are silently dropped. | ------------- | ----------------------------------------------------------------------------------------- | | SQL CLI | `./hoptimator -u "jdbc:hoptimator://hints=key1=value1,key2=value2"` | | JDBC | `props.setProperty("hints", "key1=value1,key2=value2")` | -| Subscription | `spec.hints` map on a [`Subscription`](../getting-started/concepts.md#subscriptions) CRD. | +| Subscription | `spec.hints` map on a [`Subscription`](../getting-started/concepts.md#subscriptions) custom resource. | | MCP | Set via the JDBC URL the server is launched with. (No per-call override today.) | Format: comma-separated `KEY=VALUE` pairs. URL-encode values that contain diff --git a/docs/user-guide/index.md b/docs/user-guide/index.md index 4efeb762..bc533f65 100644 --- a/docs/user-guide/index.md +++ b/docs/user-guide/index.md @@ -20,7 +20,7 @@ How to drive Hoptimator from a shell, an application, or an AI agent. ## Looking for something else -- Operating Hoptimator on Kubernetes (CRDs, configuration, the operator): +- Operating Hoptimator on Kubernetes (custom resources, configuration, the operator): [Kubernetes guide](../kubernetes/index.md) *(coming soon)*. - Adding a new database, connector, deployer, or validator: [Extending Hoptimator](../extending/index.md) *(coming soon)*. diff --git a/hoptimator-api/src/main/java/com/linkedin/hoptimator/PendingDelete.java b/hoptimator-api/src/main/java/com/linkedin/hoptimator/PendingDelete.java index 5c330aa8..e3e96fde 100644 --- a/hoptimator-api/src/main/java/com/linkedin/hoptimator/PendingDelete.java +++ b/hoptimator-api/src/main/java/com/linkedin/hoptimator/PendingDelete.java @@ -13,7 +13,7 @@ * *

An optional {@code (selfOwnerKind, selfOwnerName)} lets the caller declare an "umbrella" * K8s resource whose owned objects should be excluded from the dependent set — e.g. a - * LogicalTable CRD, so its child Pipeline CRDs (which reference tier sources by SQL) don't + * LogicalTable custom resource, so its child Pipeline custom resources (which reference tier sources by SQL) don't * self-block the drop. */ public final class PendingDelete { diff --git a/hoptimator-api/src/main/java/com/linkedin/hoptimator/Trigger.java b/hoptimator-api/src/main/java/com/linkedin/hoptimator/Trigger.java index 5bf662ef..d2337935 100644 --- a/hoptimator-api/src/main/java/com/linkedin/hoptimator/Trigger.java +++ b/hoptimator-api/src/main/java/com/linkedin/hoptimator/Trigger.java @@ -51,7 +51,7 @@ public Map options() { } /** Upstream source the trigger fires on, or {@code null} when only the name is known - * (e.g. during DROP TRIGGER / PAUSE / RESUME, which only need to look up the existing CRD). */ + * (e.g. during DROP TRIGGER / PAUSE / RESUME, which only need to look up the existing custom resource). */ public Source source() { return source; } 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 index d2ab502b..f1177387 100644 --- a/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolver.java +++ b/hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/DatabaseConfigResolver.java @@ -12,7 +12,7 @@ * 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 + * {@code hoptimator-k8s}) supplies an implementation that reads {@code Database} custom resources (or any other * registry) directly; see {@link DatabaseConfigResolverProvider}. * *

{@link #databaseProperties} deliberately mirrors {@link diff --git a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/GraphServiceTest.java b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/GraphServiceTest.java index 8274f41a..e772d2a2 100644 --- a/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/GraphServiceTest.java +++ b/hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/GraphServiceTest.java @@ -69,8 +69,8 @@ void renderUnknownFormatThrowsWithHelpfulMessage() { // ─── Identifier resolution via Calcite ──────────────────────────────────── @Test - void resolveTwoLevelResourceFindsDatabaseCrdName() throws SQLException { - // User typed ADS.AD_CLICKS. Schema ADS is a Database in the connection with CRD name ads-database, + void resolveTwoLevelResourceFindsDatabaseCrName() throws SQLException { + // User typed ADS.AD_CLICKS. Schema ADS is a Database in the connection with custom resource name ads-database, // and AD_CLICKS is a real table in its catalog (a plain physical table, not a view / LogicalTable). SchemaPlus ads = schemaWithDatabaseAndTable("ads-database", "AD_CLICKS", plainTable()); stubRootSchema(schemaWithSubs("ADS", ads)); @@ -238,20 +238,20 @@ public RelDataType getRowType(RelDataTypeFactory factory) { } /** Convenience overload: physical (non-logical) database. */ - private static SchemaPlus schemaWithDatabaseAndTable(String crdName, String tableName, Table table) { - return schemaWithDatabaseAndTable(crdName, tableName, table, /*isLogical=*/false); + private static SchemaPlus schemaWithDatabaseAndTable(String crName, String tableName, Table table) { + return schemaWithDatabaseAndTable(crName, tableName, table, /*isLogical=*/false); } /** * Mock a SchemaPlus that's (a) unwrappable as a {@link HoptimatorJdbcSchema} with the given - * CRD name and {@code isLogical()} flag, and (b) when {@code tableName != null}, exposes + * custom resource name and {@code isLogical()} flag, and (b) when {@code tableName != null}, exposes * {@code table} via {@code tables().get(tableName)}. */ - private static SchemaPlus schemaWithDatabaseAndTable(String crdName, String tableName, Table table, + private static SchemaPlus schemaWithDatabaseAndTable(String crName, String tableName, Table table, boolean isLogical) { SchemaPlus schema = schemaWithSubs(null, null); HoptimatorJdbcSchema hjs = mock(HoptimatorJdbcSchema.class); - lenient().when(hjs.databaseName()).thenReturn(crdName); + lenient().when(hjs.databaseName()).thenReturn(crName); lenient().when(hjs.isLogical()).thenReturn(isLogical); lenient().when(schema.unwrap(HoptimatorJdbcSchema.class)).thenReturn(hjs); if (tableName != null) { diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/DependencyChecker.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/DependencyChecker.java index b539b8e2..81990ba7 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/DependencyChecker.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/DependencyChecker.java @@ -18,10 +18,10 @@ /** - * Checks whether any Pipeline or TableTrigger CRDs still depend on a resource a + * Checks whether any Pipeline or TableTrigger custom resources still depend on a resource a * {@link com.linkedin.hoptimator.Deployer} is about to delete. * - *

Both CRDs carry the same {@code depends-on-} label and {@code depends-on-sources}/ + *

Both custom resources carry the same {@code depends-on-} label and {@code depends-on-sources}/ * {@code depends-on-sinks} annotations (stamped by {@link K8sPipelineDeployer} and * {@link K8sTriggerDeployer}), so the same lookup works for either: a label-selector list against * the CRD group is O(matches) on the wire, then each candidate is cross-checked against the union 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 index 28d95833..478bbf23 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolver.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolver.java @@ -15,7 +15,7 @@ /** * 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 + * {@code Database} custom resources — 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. @@ -71,7 +71,7 @@ public String databaseName(List tablePath) throws SQLException { 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 + // The database identifier is always the Database custom resource 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(). @@ -89,17 +89,17 @@ public String databaseName(List tablePath) throws SQLException { } /** - * The Database CRDs, listed once and cached for this resolver's lifetime. A resolver is built once + * The Database custom resources, 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 + *

TODO: This lists all Database custom resources 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 + * list-and-filter for now: cardinality is low (one custom resource per database) and it mirrors how * {@link K8sDatabaseTable} (the SQL/catalog path) enumerates Databases. */ private List listDatabases() throws SQLException { @@ -119,7 +119,7 @@ private List listDatabases() throws SQLException { 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 + // Catalog-style Database (e.g. MYSQL): config lives on the catalog custom resource; the requested // `schema` is a sub-schema that shares this connection. return catalog.equalsIgnoreCase(row.CATALOG); } 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 index 5b4c2ddf..66064a5c 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProvider.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverProvider.java @@ -6,7 +6,7 @@ import java.util.Properties; -/** Supplies a {@link K8sDatabaseConfigResolver} so the direct path resolves config from CRDs. */ +/** Supplies a {@link K8sDatabaseConfigResolver} so the direct path resolves config from custom resources. */ public class K8sDatabaseConfigResolverProvider implements DatabaseConfigResolverProvider { @Override diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseTable.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseTable.java index 0986e369..5d0d134a 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseTable.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sDatabaseTable.java @@ -76,7 +76,7 @@ public Row toRow(V1alpha1Database obj) { return rowOf(obj); } - /** Builds a {@link Row} from a Database CRD, usable without a {@link K8sDatabaseTable} instance. */ + /** Builds a {@link Row} from a Database custom resource, usable without a {@link K8sDatabaseTable} instance. */ static Row rowOf(V1alpha1Database obj) { return new Row(Objects.requireNonNull(obj.getMetadata()).getName(), Objects.requireNonNull(obj.getSpec()).getUrl(), obj.getSpec().getCatalog(), obj.getSpec().getSchema(), @@ -119,8 +119,8 @@ static DataSource dataSource(Row row, Properties connectionProperties) { } /** - * Builds the effective JDBC URL for a Database: its CRD {@code url} with the connection-level - * properties (except {@code user}/{@code password}) and the CRD name appended as + * Builds the effective JDBC URL for a Database: its custom resource {@code url} with the connection-level + * properties (except {@code user}/{@code password}) and the custom resource name appended as * {@code database=}. This is the URL a {@code DatabaseConfigResolver} parses to recover a * database's connection properties, kept here so it stays in lockstep with {@link #dataSource}. */ @@ -131,7 +131,7 @@ static String joinedUrl(Row row, Properties connectionProperties) { joiner.add(key + "=" + connectionProperties.getProperty(key)); } } - // Inject the Database CRD name so drivers can identify which CRD they are backing. + // Inject the Database custom resource name so drivers can identify which custom resource they are backing. // This is the value returned by source.database() in deployer/provider contexts. if (row.NAME != null && !row.NAME.isEmpty()) { joiner.add("database=" + row.NAME); diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sGraphProvider.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sGraphProvider.java index 284eb552..6d2cc6d1 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sGraphProvider.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sGraphProvider.java @@ -19,9 +19,9 @@ *

  • {@link GraphTarget.LogicalTable} → {@link PipelineGraphBuilder#forLogicalTable(String)} * (single-hop; depth is ignored — see method Javadoc) *
  • {@link GraphTarget.Resource} → {@link PipelineGraphBuilder#forResource(String, java.util.List, int)}. - * SQL-identifier-to-CRD-name resolution happens in {@link com.linkedin.hoptimator.jdbc.GraphService} + * SQL-identifier-to-custom-resource-name resolution happens in {@link com.linkedin.hoptimator.jdbc.GraphService} * before dispatch — by the time the target reaches us, {@code Resource.database()} is the - * K8s Database CRD's {@code metadata.name}. + * K8s Database custom resource's {@code metadata.name}. * * *

    Registered via {@code META-INF/services/com.linkedin.hoptimator.graph.GraphProvider} so callers diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java index ca528a88..498d194c 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java @@ -78,7 +78,7 @@ public void delete() throws SQLException { @Override public boolean exists() throws SQLException { - // The materialized view's identity is its View CRD; the Pipeline and its elements cascade from it. + // The materialized view's identity is its View custom resource; the Pipeline and its elements cascade from it. synchronized (crudLock) { return viewDeployer.exists(); } diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineBundle.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineBundle.java index b34bc1e2..d8889c70 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineBundle.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sPipelineBundle.java @@ -13,8 +13,8 @@ /** - * Deploys a Pipeline CRD together with its YAML elements (e.g. SqlJob), owned by an - * externally provided owner (such as a LogicalTable CRD rather than a View CRD). + * Deploys a Pipeline custom resource together with its YAML elements (e.g. SqlJob), owned by an + * externally provided owner (such as a LogicalTable custom resource rather than a View custom resource). * *

    Analogous to the relationship between {@link K8sMaterializedViewDeployer} and * {@link K8sViewDeployer}: just as the materialized view deployer owns a view deployer @@ -30,7 +30,7 @@ public class K8sPipelineBundle implements Deployer { /** * {@code sources} and {@code sink} are stamped as {@code depends-on-*} - * labels on the Pipeline CRD so the delete-time guard in {@link DependencyChecker} + * labels on the Pipeline custom resource so the delete-time guard in {@link DependencyChecker} * can find this pipeline by label selector. */ public K8sPipelineBundle(String name, List pipelineSpecs, String sql, @@ -63,7 +63,7 @@ public void create() throws SQLException { @Override public boolean exists() throws SQLException { - // The bundle's identity is its Pipeline CRD; the owned YAML elements cascade from it. + // The bundle's identity is its Pipeline custom resource; the owned YAML elements cascade from it. return pipelineDeployer.exists(); } @@ -79,7 +79,7 @@ public void update() throws SQLException { @Override public void delete() throws SQLException { - // Deleting the Pipeline CRD causes K8s to cascade-delete its owned YAML elements. + // Deleting the Pipeline custom resource causes K8s to cascade-delete its owned YAML elements. pipelineDeployer.delete(); } diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/LogicalTableNames.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/LogicalTableNames.java index 1d4d2121..1f9bb5a0 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/LogicalTableNames.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/LogicalTableNames.java @@ -3,7 +3,7 @@ /** * Canonical names for the K8s objects implicitly created by the LogicalTable deployer - * (inter-tier {@code Pipeline} CRDs and the offline-tier {@code TableTrigger} CRD). + * (inter-tier {@code Pipeline} custom resources and the offline-tier {@code TableTrigger} custom resource). * *

    Shared between the producer (the deployer in {@code hoptimator-logical}) and consumers * (the visualizer's {@code PipelineGraphBuilder} in {@code hoptimator-k8s}) so name-based @@ -20,12 +20,12 @@ public final class LogicalTableNames { private LogicalTableNames() { } - /** CRD name for the implicit inter-tier Pipeline between {@code fromTier} and {@code toTier}. */ + /** Custom resource name for the implicit inter-tier Pipeline between {@code fromTier} and {@code toTier}. */ public static String pipelineName(String tableName, String fromTier, String toTier) { return "logical-" + K8sUtils.canonicalizeName(tableName) + "-" + fromTier + "-to-" + toTier; } - /** CRD name for the offline-tier TableTrigger; only meaningful when an offline tier is present. */ + /** Custom resource name for the offline-tier TableTrigger; only meaningful when an offline tier is present. */ public static String triggerName(String tableName) { return String.format(TRIGGER_NAME_FORMAT, K8sUtils.canonicalizeName(tableName)); } diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilder.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilder.java index ed1988f8..6bb02b14 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilder.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilder.java @@ -36,8 +36,8 @@ /** - * Builds a {@link PipelineGraph} for visualization by walking K8s CRDs and the - * {@code depends-on-*} dependency index stamped on Pipeline CRDs. + * Builds a {@link PipelineGraph} for visualization by walking K8s custom resources and the + * {@code depends-on-*} dependency index stamped on Pipeline custom resources. * *

    Three entry points: *

      @@ -80,31 +80,31 @@ public PipelineGraphBuilder(K8sContext context) { // ─── Public entry points ───────────────────────────────────────────────── public PipelineGraph forView(String name) throws SQLException { - // Accept SQL-side identifiers (e.g. {@code VENICE.test-store$insert-partial}) — the CRD is + // Accept SQL-side identifiers (e.g. {@code VENICE.test-store$insert-partial}) — the custom resource is // stored under a canonicalized name (lowercase, {@code _} stripped, {@code $} → {@code -}, // dot-separated parts joined with {@code -}). Canonicalization is idempotent, so passing the - // already-canonical CRD name still works. - String crdName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\."))); - V1alpha1View view = viewApi.get(crdName); + // already-canonical custom resource name still works. + String crName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\."))); + V1alpha1View view = viewApi.get(crName); if (view.getSpec() == null) { - throw new SQLException("view " + crdName + " not found"); + throw new SQLException("view " + crName + " not found"); } boolean materialized = Boolean.TRUE.equals(view.getSpec().getMaterialized()); - GraphNode.View root = new GraphNode.View(crdName, materialized); + GraphNode.View root = new GraphNode.View(crName, materialized); Traversal t = new Traversal(); t.addNode(root); - // Materialized views own a Pipeline with the same CRD name (see K8sMaterializedViewDeployer). + // Materialized views own a Pipeline with the same custom resource name (see K8sMaterializedViewDeployer). // Look it up directly — avoids a namespace-wide LIST. The owner-ref check still runs to defend // against a coincidentally-named pipeline that isn't actually owned by this view (e.g. stale - // pipeline from a recreated view CRD with a fresh UID). Expansion is single-hop: "what this + // pipeline from a recreated view custom resource with a fresh UID). Expansion is single-hop: "what this // view does," not the full upstream chain — for that, run !graph on a source identifier // (Resource targets honor depth). if (materialized) { String viewUid = view.getMetadata() == null ? null : view.getMetadata().getUid(); - V1alpha1Pipeline pipeline = pipelineApi.getIfExists(crdName); - if (pipeline != null && ownedBy(pipeline.getMetadata(), "View", crdName, viewUid)) { + V1alpha1Pipeline pipeline = pipelineApi.getIfExists(crName); + if (pipeline != null && ownedBy(pipeline.getMetadata(), "View", crName, viewUid)) { GraphNode.Pipeline pipeNode = t.expandPipelineDirected(pipeline, Direction.UPSTREAM, 0); if (pipeNode != null) { t.addEdge(new GraphEdge(root, pipeNode, GraphEdge.Type.OWNER_OF)); @@ -117,11 +117,11 @@ public PipelineGraph forView(String name) throws SQLException { public PipelineGraph forLogicalTable(String name) throws SQLException { // Same canonicalization as forView — accept SQL-side identifiers and resolve to the - // canonicalized CRD name. - String crdName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\."))); - V1alpha1LogicalTable lt = logicalTableApi.get(crdName); + // canonicalized custom resource name. + String crName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\."))); + V1alpha1LogicalTable lt = logicalTableApi.get(crName); Map tierMap = tierMap(lt.getSpec()); - GraphNode.LogicalTable root = new GraphNode.LogicalTable(crdName, tierMap); + GraphNode.LogicalTable root = new GraphNode.LogicalTable(crName, tierMap); Traversal t = new Traversal(); t.addNode(root); @@ -135,7 +135,7 @@ public PipelineGraph forLogicalTable(String name) throws SQLException { // nearline→online, nearline→offline), so most candidate GETs miss with 404 — that's fine, and // it keeps the visualizer decoupled from the deployer's pair-selection rules: whatever the // deployer actually creates is what we render. Owner-ref check defends against a coincidentally - // -named pipeline owned by something else (or a stale CRD with a different UID). Expansion is + // -named pipeline owned by something else (or a stale custom resource with a different UID). Expansion is // single-hop — for the chain view, run !graph on a source identifier (Resource targets honor // depth). if (tableName != null) { @@ -147,7 +147,7 @@ public PipelineGraph forLogicalTable(String name) throws SQLException { } String candidate = LogicalTableNames.pipelineName(tableName, fromTier, toTier); V1alpha1Pipeline pipeline = pipelineApi.getIfExists(candidate); - if (pipeline != null && ownedBy(pipeline.getMetadata(), "LogicalTable", crdName, ltUid)) { + if (pipeline != null && ownedBy(pipeline.getMetadata(), "LogicalTable", crName, ltUid)) { GraphNode.Pipeline pipeNode = t.expandPipelineDirected(pipeline, Direction.UPSTREAM, 0); if (pipeNode != null) { t.addEdge(new GraphEdge(root, pipeNode, GraphEdge.Type.OWNER_OF)); @@ -175,7 +175,7 @@ public PipelineGraph forLogicalTable(String name) throws SQLException { if (tableName != null) { String triggerCandidate = LogicalTableNames.triggerName(tableName); V1alpha1TableTrigger trigger = triggerApi.getIfExists(triggerCandidate); - if (trigger != null && ownedBy(trigger.getMetadata(), "LogicalTable", crdName, ltUid)) { + if (trigger != null && ownedBy(trigger.getMetadata(), "LogicalTable", crName, ltUid)) { GraphNode.Trigger tNode = triggerNode(trigger); t.addNode(tNode); t.addEdge(new GraphEdge(root, tNode, GraphEdge.Type.OWNER_OF)); @@ -566,7 +566,7 @@ void expandFromResource(GraphNode.External resource, int remainingDepth, Directi } /** - * Ingest a Pipeline CRD into the graph. Reads source/sink annotations, creates edges, + * Ingest a Pipeline custom resource into the graph. Reads source/sink annotations, creates edges, * and schedules transitive expansion from the other endpoints. Returns the Pipeline node * so the caller can attach owner-of edges. */ diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1LogicalTableSpecTiers.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1LogicalTableSpecTiers.java index 52150cad..3677b717 100644 --- a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1LogicalTableSpecTiers.java +++ b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1LogicalTableSpecTiers.java @@ -41,10 +41,10 @@ public V1alpha1LogicalTableSpecTiers database(String database) { } /** - * Name of the Database CRD backing this tier. + * Name of the Database custom resource backing this tier. * @return database **/ - @ApiModelProperty(required = true, value = "Name of the Database CRD backing this tier.") + @ApiModelProperty(required = true, value = "Name of the Database custom resource backing this tier.") public String getDatabase() { return database; diff --git a/hoptimator-k8s/src/main/resources/logicaltables.crd.yaml b/hoptimator-k8s/src/main/resources/logicaltables.crd.yaml index 491c11d6..3aaf12f3 100644 --- a/hoptimator-k8s/src/main/resources/logicaltables.crd.yaml +++ b/hoptimator-k8s/src/main/resources/logicaltables.crd.yaml @@ -42,7 +42,7 @@ spec: type: object properties: database: - description: Name of the Database CRD backing this tier. + description: Name of the Database custom resource backing this tier. type: string required: - database 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 index 8395a5b0..ce50891b 100644 --- a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverTest.java +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sDatabaseConfigResolverTest.java @@ -91,7 +91,7 @@ void databasePropertiesParsesUrlForSchemaStyleDatabase() throws Exception { assertThat(props).isNotNull(); assertThat(props.getProperty("bootstrap.servers")).isEqualTo("localhost:9092"); - // joinedUrl injects the CRD name as database=. + // joinedUrl injects the custom resource name as database=. assertThat(props.getProperty("database")).isEqualTo("kafka-database"); } @@ -109,7 +109,7 @@ void databasePropertiesReturnsNullWhenNoDatabaseMatches() throws Exception { } @Test - void databaseNameReturnsCrdNameForSchemaStyle() throws Exception { + void databaseNameReturnsCrNameForSchemaStyle() throws Exception { V1alpha1Database kafka = db("kafka-database", "jdbc:kafka://bootstrap.servers=localhost:9092", null, "KAFKA"); @@ -121,7 +121,7 @@ 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. + // Catalog-style: three-segment path [CATALOG, SCHEMA, TABLE] matches on the catalog custom resource. assertThat(resolver(mysql).databaseName(Arrays.asList("MYSQL", "test_database", "orders"))) .isEqualTo("mysql"); } @@ -191,7 +191,7 @@ void listsDatabasesOncePerResolverAcrossMultipleResolutions() throws Exception { resolver.databaseProperties(null, "KAFKA", "jdbc:kafka://"); resolver.databaseName(Arrays.asList("KAFKA", "t2")); - // The per-resolver cache means the Database CRDs are listed only once. + // The per-resolver cache means the Database custom resources 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/K8sTriggerDeployerTest.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTriggerDeployerTest.java index b0713f12..370b739b 100644 --- a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTriggerDeployerTest.java +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sTriggerDeployerTest.java @@ -426,7 +426,7 @@ void deleteOnNonExistingTriggerThrowsSqlException() { @Test void updatePreservesPausedWhenOptionsHaveNoPausedOption() throws SQLException { // Scenario: CREATE OR REPLACE TABLE with no explicit pause/resume — an existing paused - // CRD must remain paused even though the caller passes no PAUSED_OPTION. + // custom resource must remain paused even though the caller passes no PAUSED_OPTION. V1alpha1TableTrigger existing = new V1alpha1TableTrigger() .metadata(new V1ObjectMeta().name("mytrigger")) .spec(new V1alpha1TableTriggerSpec().paused(true)); @@ -444,7 +444,7 @@ void updatePreservesPausedWhenOptionsHaveNoPausedOption() throws SQLException { @Test void updatePreservesUnpausedWhenOptionsHaveNoPausedOption() throws SQLException { - // Scenario: existing CRD is unpaused and the caller passes no PAUSED_OPTION — stays unpaused. + // Scenario: existing custom resource is unpaused and the caller passes no PAUSED_OPTION — stays unpaused. V1alpha1TableTrigger existing = new V1alpha1TableTrigger() .metadata(new V1ObjectMeta().name("mytrigger")) .spec(new V1alpha1TableTriggerSpec().paused(false)); @@ -477,7 +477,7 @@ void updateWithPausedOptionFalseUnpausesAlreadyPausedTrigger() throws SQLExcepti deployer.update(); assertFalse(triggers.get(0).getSpec().getPaused(), - "Explicit PAUSED_OPTION=false must override a currently-paused CRD"); + "Explicit PAUSED_OPTION=false must override a currently-paused custom resource"); } @Test @@ -502,7 +502,7 @@ void updateFallsThroughToSuperUpdateWhenNoExistingAndNoPausedOption() throws SQL @Test void updateFallsThroughToSuperUpdateWhenExistingHasNullPaused() throws SQLException { - // Scenario: existing CRD has spec.paused=null — neither the explicit-option nor the + // Scenario: existing custom resource has spec.paused=null — neither the explicit-option nor the // preserve-existing path fires, so super.update() re-renders (which calls toK8sObject, // which needs a JobTemplate). V1alpha1TableTrigger existing = new V1alpha1TableTrigger() @@ -521,10 +521,10 @@ void updateFallsThroughToSuperUpdateWhenExistingHasNullPaused() throws SQLExcept K8sTriggerDeployer deployer = makeDeployer(trigger, mockContext); deployer.update(); - // If super.update() ran, toK8sObject() populated spec.table on the re-rendered CRD. + // If super.update() ran, toK8sObject() populated spec.table on the re-rendered custom resource. // The short-circuit would instead leave spec.table untouched (null on the existing). assertTrue(triggers.stream().anyMatch(t -> "TABLE".equals(t.getSpec().getTable())), - "super.update() must have run toK8sObject() — expected a CRD with spec.table='TABLE'"); + "super.update() must have run toK8sObject() — expected a custom resource with spec.table='TABLE'"); } // ───────── toK8sObject() catalog rendering tests ───────── diff --git a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilderTest.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilderTest.java index e9f26dc2..679b5078 100644 --- a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilderTest.java +++ b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilderTest.java @@ -35,7 +35,7 @@ /** - * End-to-end tests for {@link PipelineGraphBuilder}. Each test seeds a small fixture of CRDs + * End-to-end tests for {@link PipelineGraphBuilder}. Each test seeds a small fixture of custom resources * via {@link FakeK8sApi} and exercises one of the three entry points. Assertions probe graph * structure (node kinds, edge types, ownership wiring) rather than rendered output, which is * tested separately in {@code MermaidRendererTest}. @@ -47,7 +47,7 @@ class PipelineGraphBuilderTest { @Test void forViewMaterializedRendersSourcesAndSink() throws SQLException { V1alpha1View view = view("ads", "audience", true, "view-uid-1"); - // Pipeline name == view CRD name (see K8sMaterializedViewDeployer); the visualizer derives + // Pipeline name == view custom resource name (see K8sMaterializedViewDeployer); the visualizer derives // the name from the view, so the fixture must mirror that contract. V1alpha1Pipeline pipeline = pipelineWithSourcesAndSink( "audience", "view-uid-1", "View", @@ -299,7 +299,7 @@ void forLogicalTableIncludesPipelineViaOwnerNameFallback() throws SQLException { @Test void forLogicalTableSkipsPipelineWithWrongOwnerKind() throws SQLException { // A pipeline coincidentally named like a LogicalTable's inter-tier pipeline but actually - // owned by a different kind (e.g. a Job CRD) must NOT be claimed as the LogicalTable's child. + // owned by a different kind (e.g. a Job custom resource) must NOT be claimed as the LogicalTable's child. // ownedBy()'s kind check is the guard. V1alpha1LogicalTable lt = logicalTable("ns", "events", "lt-uid", linked( "nearline", "kafka-db", @@ -325,7 +325,7 @@ void forLogicalTableSkipsPipelineWithWrongOwnerKind() throws SQLException { @Test void forLogicalTableHandlesMissingUidAndSpec() throws SQLException { - // CRD shapes pulled from K8s can have a partially-populated metadata block (e.g. missing + // Custom resource shapes pulled from K8s can have a partially-populated metadata block (e.g. missing // UID before the apiserver assigns one) or a null spec mid-reconciliation. The visualizer // must surface the LogicalTable root rather than NPE'ing on the missing fields. Owned // children won't be discoverable (no UID for owner-ref check, no tableName for name @@ -483,7 +483,7 @@ void largeDepthHandledGracefully() throws SQLException { @Test void forViewCanonicalizesSqlStyleIdentifier() throws SQLException { - // CRD is stored as `venice-test-store-insert-partial`. The user passes the SQL-side identifier + // Custom resource is stored as `venice-test-store-insert-partial`. The user passes the SQL-side identifier // they typed in their CREATE statement, which has uppercase, dots, and a `$`. V1alpha1View view = view("ns", "venice-test-store-insert-partial", true, "uid-v"); diff --git a/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java b/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java index 879ac666..c174b028 100644 --- a/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java +++ b/hoptimator-kafka/src/main/java/com/linkedin/hoptimator/kafka/KafkaDeployerProvider.java @@ -19,7 +19,7 @@ *

      Detection uses {@link DeploymentContext#databaseProperties} to resolve the source's * {@code Database} config and checks whether its connection URL starts with {@code jdbc:kafka://}. * This is Calcite-free: it works identically whether the config comes from the Calcite catalog - * (SQL path) or from a {@code Database} CRD (direct path). The Kafka config (bootstrap.servers) is + * (SQL path) or from a {@code Database} custom resource (direct path). The Kafka config (bootstrap.servers) is * parsed from that URL. */ public class KafkaDeployerProvider implements DeployerProvider { diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployer.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployer.java index e014a851..e958d257 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployer.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployer.java @@ -14,22 +14,22 @@ /** - * Deploys a {@code LogicalTable} CRD, analogous to {@link com.linkedin.hoptimator.k8s.K8sPipelineDeployer} - * for Pipeline CRDs. Inherits snapshot-based restore from {@link K8sDeployer}: if the CRD did not + * Deploys a {@code LogicalTable} custom resource, analogous to {@link com.linkedin.hoptimator.k8s.K8sPipelineDeployer} + * for Pipeline custom resources. Inherits snapshot-based restore from {@link K8sDeployer}: if the custom resource did not * exist before this deployer ran, restore() deletes it; if it existed, restore() reverts it to its * prior state. */ class K8sLogicalTableDeployer extends K8sDeployer { - private final String crdName; + private final String crName; private final String databaseLabel; private final String tableName; private final Map tierMap; - K8sLogicalTableDeployer(String crdName, String databaseLabel, String tableName, + K8sLogicalTableDeployer(String crName, String databaseLabel, String tableName, Map tierMap, K8sContext context) { super(context, K8sApiEndpoints.LOGICAL_TABLES); - this.crdName = crdName; + this.crName = crName; this.databaseLabel = databaseLabel; this.tableName = tableName; this.tierMap = tierMap; @@ -46,7 +46,7 @@ protected V1alpha1LogicalTable toK8sObject() { return new V1alpha1LogicalTable() .kind(K8sApiEndpoints.LOGICAL_TABLES.kind()) .apiVersion(K8sApiEndpoints.LOGICAL_TABLES.apiVersion()) - .metadata(new V1ObjectMeta().name(crdName) + .metadata(new V1ObjectMeta().name(crName) .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, databaseLabel)) .spec(spec); } diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTable.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTable.java index 5d1c313f..60447f7a 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTable.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTable.java @@ -26,7 +26,7 @@ /** - * A Calcite table backed by a {@code LogicalTable} CRD. + * A Calcite table backed by a {@code LogicalTable} custom resource. * *

      The row type is resolved lazily on first access by opening a JDBC connection to * the preferred physical tier (nearline first) and reading the schema from there. @@ -105,12 +105,12 @@ private RelDataType resolveRowType() { String databaseName = tierBinding.getDatabase(); try { - V1alpha1Database dbCrd = databaseApi.get(databaseName); - if (dbCrd.getSpec() == null) { + V1alpha1Database dbCr = databaseApi.get(databaseName); + if (dbCr.getSpec() == null) { return null; } - String tierUrl = dbCrd.getSpec().getUrl(); - String tierSchema = dbCrd.getSpec().getSchema(); + String tierUrl = dbCr.getSpec().getUrl(); + String tierSchema = dbCr.getSpec().getSchema(); try (Connection tierConn = DriverManager.getConnection(tierUrl)) { CalciteConnection calciteConn = tierConn.unwrap(CalciteConnection.class); 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 81d508fb..15740976 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 @@ -62,8 +62,8 @@ *

        *
      1. Validating that at least two tiers are defined.
        2. *
      2. Creating physical resources for each tier via the tier-specific deployer SPI.
        3. - *
      3. Creating a {@code LogicalTable} CRD via {@link K8sLogicalTableDeployer}.
        4. - *
      4. Creating implicit inter-tier Pipeline CRDs owned by the LogicalTable CRD.
        5. + *
      5. Creating a {@code LogicalTable} custom resource via {@link K8sLogicalTableDeployer}.
        6. + *
      6. Creating implicit inter-tier Pipeline custom resources owned by the LogicalTable custom resource.
        7. *
      * * TODO: this deployer is specific to k8s at this point in time, this should be refactored to be @@ -83,7 +83,7 @@ public class LogicalTableDeployer implements Deployer, Validated { private final List triggerDeployers = new ArrayList<>(); private final List schemaRollbacks = new ArrayList<>(); - // Created lazily in deployAll(). Null until then, so restore() is a no-op for the CRD if + // Created lazily in deployAll(). Null until then, so restore() is a no-op for the custom resource if // deployAll() never ran (e.g. validation failed before deployment started). private K8sLogicalTableDeployer logicalTableDeployer; @@ -106,14 +106,14 @@ public class LogicalTableDeployer implements Deployer, Validated { } /** - * Factory method for the LogicalTable CRD deployer — overridable in tests. + * Factory method for the LogicalTable custom resource deployer — overridable in tests. * The deployer inherits snapshot-based restore from {@link com.linkedin.hoptimator.k8s.K8sDeployer}: - * if the CRD didn't exist before this deployment it is deleted on restore(); if it existed it + * if the custom resource didn't exist before this deployment it is deleted on restore(); if it existed it * is reverted to its prior state. */ K8sLogicalTableDeployer createLogicalTableDeployer( - String crdName, String databaseLabel, Map tierMap) { - return new K8sLogicalTableDeployer(crdName, databaseLabel, source.table(), tierMap, context); + String crName, String databaseLabel, Map tierMap) { + return new K8sLogicalTableDeployer(crName, databaseLabel, source.table(), tierMap, context); } /** @@ -186,7 +186,7 @@ public void create() throws SQLException { @Override public boolean exists() throws SQLException { - // A logical table's identity is its LogicalTable CRD (the implicit inter-tier pipelines and + // A logical table's identity is its LogicalTable custom resource (the implicit inter-tier pipelines and // triggers are owned by it and cascade). Mirror how delete() names and builds that deployer. String selfName = K8sUtils.canonicalizeName(source.path()); return createLogicalTableDeployer(selfName, source.database(), buildTierMap()).exists(); @@ -199,7 +199,7 @@ public void update() throws SQLException { /** * Core create/update logic. Tier resources are always deployed with update() semantics - * (create-or-update). The LogicalTable CRD is managed by {@link K8sLogicalTableDeployer} + * (create-or-update). The LogicalTable custom resource is managed by {@link K8sLogicalTableDeployer} * which snapshots the prior state before acting, enabling correct restore() behavior. */ private void deployAll(boolean update) throws SQLException { @@ -208,8 +208,8 @@ private void deployAll(boolean update) throws SQLException { try { deployTierResources(tierSources); - String crdName = K8sUtils.canonicalizeName(source.path()); - logicalTableDeployer = createLogicalTableDeployer(crdName, source.database(), tierMap); + String crName = K8sUtils.canonicalizeName(source.path()); + logicalTableDeployer = createLogicalTableDeployer(crName, source.database(), tierMap); V1OwnerReference ownerRef = update ? logicalTableDeployer.updateAndReference() : logicalTableDeployer.createAndReference(); @@ -229,15 +229,15 @@ private void deployAll(boolean update) throws SQLException { * Deletes a logical table. * *

      A logical DROP is structurally equivalent to running DROP TABLE on each tier plus - * deleting the LogicalTable CRD. We mirror that shape exactly: each tier goes through the + * deleting the LogicalTable custom resource. We mirror that shape exactly: each tier goes through the * same {@code validateOrThrow → DeploymentService.delete} pipeline a standalone DROP would. * The {@link PendingDelete}'s {@code (selfOwnerKind, selfOwnerName)} pair identifies the - * LogicalTable CRD so the implicit inter-tier pipelines (owned by the CRD, cascade-deleted + * LogicalTable custom resource so the implicit inter-tier pipelines (owned by the custom resource, cascade-deleted * with it) are excluded from the dependent set — only external pipelines block. * *

        *
      1. Per-tier dep check via the validator framework. Any active external pipeline blocks. - *
      2. Delete the {@code LogicalTable} CRD. K8s owner-ref cascade removes its implicit + *
      3. Delete the {@code LogicalTable} custom resource. K8s owner-ref cascade removes its implicit * inter-tier Pipelines and their Flink/YAML children. Must succeed. *
      4. Per-tier physical cleanup (Kafka topic, Venice store, ...). Best effort — a * stranded tier resource is recoverable; aborting mid-DROP isn't. @@ -255,7 +255,7 @@ public void delete() throws SQLException { new PendingDelete<>(tierSource, "LogicalTable", selfName), context.deploymentContext()); } - // 2. Delete the LogicalTable CRD (cascades owned pipelines/triggers). + // 2. Delete the LogicalTable custom resource (cascades owned pipelines/triggers). createLogicalTableDeployer(selfName, source.database(), buildTierMap()).delete(); // 3. Per-tier physical cleanup. Best-effort: only deregister a tier's schema entry when its @@ -282,15 +282,15 @@ public void delete() throws SQLException { @Override public void restore() { - // Restore trigger deployers first (they reference the LogicalTable CRD via ownerRef) + // Restore trigger deployers first (they reference the LogicalTable custom resource via ownerRef) for (int i = triggerDeployers.size() - 1; i >= 0; i--) { triggerDeployers.get(i).restore(); } - // Restore pipeline deployers next (they also reference the LogicalTable CRD via ownerRef) + // Restore pipeline deployers next (they also reference the LogicalTable custom resource via ownerRef) for (int i = pipelineDeployers.size() - 1; i >= 0; i--) { pipelineDeployers.get(i).restore(); } - // Delegate CRD rollback to the deployer: if it didn't exist before, it's deleted; + // Delegate custom resource rollback to the deployer: if it didn't exist before, it's deleted; // if it existed, it's reverted to its prior state; if deployAll() never ran it's null. if (logicalTableDeployer != null) { logicalTableDeployer.restore(); @@ -442,7 +442,7 @@ Map buildTierMap() { } /** - * Resolves each tier's Database CRD in a single pass. + * Resolves each tier's Database custom resource in a single pass. */ private Map resolveTiers() throws SQLException { if (cachedTierDatabases == null) { @@ -469,7 +469,7 @@ private void deployTierResources(Map tierSources) throws SQLExce for (Map.Entry entry : tierSources.entrySet()) { String tierName = entry.getKey(); Source tierSource = entry.getValue(); - log.info("Deploying tier {} (database CRD: {}) for table {}", + log.info("Deploying tier {} (database custom resource: {}) for table {}", tierName, tierSource.database(), source.table()); Collection deployers = DeploymentService.deployers(tierSource, context.deploymentContext()); for (Deployer deployer : deployers) { @@ -482,7 +482,7 @@ private void deployTierResources(Map tierSources) throws SQLExce } /** - * Creates or updates implicit inter-tier pipeline CRDs. + * Creates or updates implicit inter-tier pipeline custom resources. * Each created bundle is tracked in {@link #pipelineDeployers}. */ private void deployImplicitPipelines(Map tierMap, Map tierSources, @@ -545,7 +545,7 @@ static String pipelineName(String tableName, String fromTier, String toTier) { /** * Creates or updates a {@link V1alpha1TableTrigger} for the offline tier, owned by the - * LogicalTable CRD. + * LogicalTable custom resource. */ private void deployImplicitTrigger(Map tierMap, Map tierSources, K8sContext ownerContext) throws SQLException { diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployerProvider.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployerProvider.java index 4facd25f..919a93c8 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployerProvider.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployerProvider.java @@ -18,7 +18,7 @@ /** - * Activates {@link LogicalTableDeployer} for sources backed by a logical Database CRD. + * Activates {@link LogicalTableDeployer} for sources backed by a logical Database custom resource. * *

        Detection uses {@link DeploymentContext#databaseProperties} to look up the * schema by name and check if its underlying JDBC URL starts with diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDriver.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDriver.java index 72a227a2..39dcae22 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDriver.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDriver.java @@ -25,14 +25,14 @@ * *

        The schema name under which the {@link LogicalTableSchema} is registered must be * provided via the {@code schema} connection property (set by the operator when it - * reads the Database CRD). If not provided, it defaults to {@code "LOGICAL"}. + * reads the Database custom resource). If not provided, it defaults to {@code "LOGICAL"}. */ public class LogicalTableDriver extends CalciteDriver { public static final String CONNECT_STRING_PREFIX = "jdbc:logical://"; /** Connection property that hints which tier to resolve for pipeline planning (e.g. "nearline", "online"). */ public static final String TIER_PROPERTY = "tier"; - /** K8s label key identifying which logical database a LogicalTable CRD belongs to. */ + /** K8s label key identifying which logical database a LogicalTable custom resource belongs to. */ public static final String DATABASE_LABEL = "logical-database"; static { @@ -69,7 +69,7 @@ public Connection connect(String url, Properties props) throws SQLException { + tierCount + " tier(s) found in: " + url); } - // The database name (Database CRD metadata.name) is injected by K8sDatabaseTable. + // The database name (Database custom resource metadata.name) is injected by K8sDatabaseTable. // It is used as the label filter in LogicalTableSchema and matches source.database() // in deployer/provider contexts. The Calcite schema registration name is handled by // the outer catalog (K8sDatabaseTable.addDatabases) — not by this driver. @@ -77,7 +77,7 @@ public Connection connect(String url, Properties props) throws SQLException { if (databaseName == null || databaseName.isEmpty()) { throw new SQLNonTransientException( "Missing 'database' property in logical database URL. " - + "Ensure the Database CRD has metadata.name set (injected by K8sDatabaseTable)."); + + "Ensure the Database custom resource has metadata.name set (injected by K8sDatabaseTable)."); } try { diff --git a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableSchema.java b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableSchema.java index b67cbdc4..46d0feee 100644 --- a/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableSchema.java +++ b/hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableSchema.java @@ -26,9 +26,9 @@ /** - * A Calcite schema that lists {@link LogicalTable} instances from K8s CRDs. + * A Calcite schema that lists {@link LogicalTable} instances from K8s custom resources. * - *

        CRDs are filtered by the {@link LogicalTableDriver#DATABASE_LABEL} label, which the deployer + *

        Custom resources are filtered by the {@link LogicalTableDriver#DATABASE_LABEL} label, which the deployer * sets to the database/schema name at creation time. Row type resolution is performed * lazily inside each {@link LogicalTable} on first access — not eagerly here. */ @@ -67,12 +67,12 @@ protected Map loadAll() throws Exception { @Override @Nullable protected Table load(String name) throws Exception { - // Direct CRD lookup by name avoids a full list scan for single-table access. + // Direct custom resource lookup by name avoids a full list scan for single-table access. // getIfExists() returns null on 404 (table not found) and throws SQLException // for any other K8s error (auth, network, server error), which propagates here. - String expectedCrdName = K8sUtils.canonicalizeName(databaseName, name); - V1alpha1LogicalTable crd = logicalTableApi.getIfExists(context.namespace(), expectedCrdName); - return crd != null ? tableFromCrd(crd) : null; + String expectedCrName = K8sUtils.canonicalizeName(databaseName, name); + V1alpha1LogicalTable cr = logicalTableApi.getIfExists(context.namespace(), expectedCrName); + return cr != null ? tableFromCr(cr) : null; } @Override @@ -93,15 +93,15 @@ protected Map getTableMap() { } private Map loadTableMap() throws SQLException { - Collection crds = logicalTableApi.list(); + Collection crs = logicalTableApi.list(); Map result = new HashMap<>(); - for (V1alpha1LogicalTable crd : crds) { - Table table = tableFromCrd(crd); + for (V1alpha1LogicalTable cr : crs) { + Table table = tableFromCr(cr); if (table == null) { continue; } - // Use spec.tableName (the original declared name) not the K8s CRD metadata.name + // Use spec.tableName (the original declared name) not the K8s custom resource metadata.name // (which is a compound "database-table" key for K8s uniqueness). result.put(((LogicalTable) table).name(), table); } @@ -110,22 +110,22 @@ private Map loadTableMap() throws SQLException { } @Nullable - Table tableFromCrd(V1alpha1LogicalTable crd) { - if (crd.getMetadata() == null || crd.getSpec() == null) { + Table tableFromCr(V1alpha1LogicalTable cr) { + if (cr.getMetadata() == null || cr.getSpec() == null) { return null; } - Map labels = crd.getMetadata().getLabels(); + Map labels = cr.getMetadata().getLabels(); String label = labels != null ? labels.get(LogicalTableDriver.DATABASE_LABEL) : null; if (!databaseName.equalsIgnoreCase(label)) { return null; } - if (crd.getSpec().getTiers() == null || crd.getSpec().getTiers().isEmpty()) { - log.warn("LogicalTable CRD {} has no tier bindings; skipping", crd.getMetadata().getName()); + if (cr.getSpec().getTiers() == null || cr.getSpec().getTiers().isEmpty()) { + log.warn("LogicalTable custom resource {} has no tier bindings; skipping", cr.getMetadata().getName()); return null; } // spec.tableName is the original declared name; metadata.name is the compound K8s resource name. - String tableName = crd.getSpec().getTableName() != null - ? crd.getSpec().getTableName() : crd.getMetadata().getName(); - return new LogicalTable(tableName, crd.getSpec().getTiers(), resolvedTier, context); + String tableName = cr.getSpec().getTableName() != null + ? cr.getSpec().getTableName() : cr.getMetadata().getName(); + return new LogicalTable(tableName, cr.getSpec().getTiers(), resolvedTier, context); } } diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployerTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployerTest.java index 0e08999d..9c76cd5e 100644 --- a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployerTest.java +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/K8sLogicalTableDeployerTest.java @@ -13,9 +13,9 @@ /** * Unit tests for {@link K8sLogicalTableDeployer#toK8sObject()}. * - *

        These tests verify the structure of the {@code LogicalTable} CRD object built by the + *

        These tests verify the structure of the {@code LogicalTable} custom resource object built by the * deployer — metadata (name, DATABASE_LABEL), spec (tableName, tiers map). They complement - * {@link LogicalTableDeployerTest} which now mocks the CRD deployer and delegates content + * {@link LogicalTableDeployerTest} which now mocks the custom resource deployer and delegates content * checks here. */ class K8sLogicalTableDeployerTest { @@ -26,9 +26,9 @@ void toK8sObjectHasCorrectName() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-testevent", "logical", "testevent", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals("logical-testevent", crd.getMetadata().getName()); + assertEquals("logical-testevent", cr.getMetadata().getName()); } @Test @@ -37,11 +37,11 @@ void toK8sObjectHasDatabaseLabel() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-myevent", "logical", "myevent", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertNotNull(crd.getMetadata().getLabels()); + assertNotNull(cr.getMetadata().getLabels()); assertEquals("logical", - crd.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); + cr.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); } @Test @@ -50,9 +50,9 @@ void toK8sObjectHasCorrectTableName() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-testevent", "logical", "testevent", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals("testevent", crd.getSpec().getTableName()); + assertEquals("testevent", cr.getSpec().getTableName()); } @Test @@ -61,11 +61,11 @@ void toK8sObjectHasCorrectTiersMap() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-testevent", "logical", "testevent", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals(2, crd.getSpec().getTiers().size()); - assertEquals("kafka-db", crd.getSpec().getTiers().get("nearline").getDatabase()); - assertEquals("openhouse-db", crd.getSpec().getTiers().get("offline").getDatabase()); + assertEquals(2, cr.getSpec().getTiers().size()); + assertEquals("kafka-db", cr.getSpec().getTiers().get("nearline").getDatabase()); + assertEquals("openhouse-db", cr.getSpec().getTiers().get("offline").getDatabase()); } @Test @@ -74,14 +74,14 @@ void toK8sObjectHasCorrectMetadata() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-testevent", "logical", "testevent", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals("logical-testevent", crd.getMetadata().getName()); + assertEquals("logical-testevent", cr.getMetadata().getName()); assertEquals("logical", - crd.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); - assertEquals("testevent", crd.getSpec().getTableName()); - assertEquals(2, crd.getSpec().getTiers().size()); - assertEquals("kafka-db", crd.getSpec().getTiers().get("nearline").getDatabase()); + cr.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); + assertEquals("testevent", cr.getSpec().getTableName()); + assertEquals(2, cr.getSpec().getTiers().size()); + assertEquals("kafka-db", cr.getSpec().getTiers().get("nearline").getDatabase()); } @Test @@ -90,13 +90,13 @@ void toK8sObjectWithSingleTier() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-orders", "mydb", "orders", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals("logical-orders", crd.getMetadata().getName()); - assertEquals("mydb", crd.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); - assertEquals("orders", crd.getSpec().getTableName()); - assertEquals(1, crd.getSpec().getTiers().size()); - assertEquals("venice-db", crd.getSpec().getTiers().get("online").getDatabase()); + assertEquals("logical-orders", cr.getMetadata().getName()); + assertEquals("mydb", cr.getMetadata().getLabels().get(LogicalTableDriver.DATABASE_LABEL)); + assertEquals("orders", cr.getSpec().getTableName()); + assertEquals(1, cr.getSpec().getTiers().size()); + assertEquals("venice-db", cr.getSpec().getTiers().get("online").getDatabase()); } @Test @@ -108,11 +108,11 @@ void toK8sObjectWithThreeTiers() { K8sLogicalTableDeployer deployer = new K8sLogicalTableDeployer( "logical-events", "mydb", "events", tierMap, null); - V1alpha1LogicalTable crd = deployer.toK8sObject(); + V1alpha1LogicalTable cr = deployer.toK8sObject(); - assertEquals(3, crd.getSpec().getTiers().size()); - assertEquals("kafka-db", crd.getSpec().getTiers().get("nearline").getDatabase()); - assertEquals("openhouse-db", crd.getSpec().getTiers().get("offline").getDatabase()); - assertEquals("venice-db", crd.getSpec().getTiers().get("online").getDatabase()); + assertEquals(3, cr.getSpec().getTiers().size()); + assertEquals("kafka-db", cr.getSpec().getTiers().get("nearline").getDatabase()); + assertEquals("openhouse-db", cr.getSpec().getTiers().get("offline").getDatabase()); + assertEquals("venice-db", cr.getSpec().getTiers().get("online").getDatabase()); } } diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableDeployerTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableDeployerTest.java index 55f0de5f..2fe0b538 100644 --- a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableDeployerTest.java +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableDeployerTest.java @@ -96,7 +96,7 @@ class LogicalTableDeployerTest { MockedStatic validationServiceMock; @Mock - K8sLogicalTableDeployer mockCrdDeployer; + K8sLogicalTableDeployer mockCrDeployer; // Helper methods shared by outer class tests @@ -144,24 +144,24 @@ private static Source testSource() { } /** - * Creates a {@link LogicalTableDeployer} that returns {@link #mockCrdDeployer} from its - * {@code createLogicalTableDeployer} factory method, so K8s CRD interactions can be + * Creates a {@link LogicalTableDeployer} that returns {@link #mockCrDeployer} from its + * {@code createLogicalTableDeployer} factory method, so K8s custom resource interactions can be * verified with Mockito without a live cluster. */ - private LogicalTableDeployer deployerWithMockCrd( + private LogicalTableDeployer deployerWithMockCr( Source src, Properties props, K8sContext ctx, FakeK8sApi dbApi) { return new LogicalTableDeployer(src, props, ctx, dbApi) { @Override K8sLogicalTableDeployer createLogicalTableDeployer( - String crdName, String databaseLabel, Map tierMap) { - return mockCrdDeployer; + String crName, String databaseLabel, Map tierMap) { + return mockCrDeployer; } @Override void deployPipelineBundle(String fromTier, String toTier, Map tierSources, K8sContext ownerContext, boolean update) { - // No-op in CRD-focused tests — pipeline deployment is tested separately. + // No-op in custom-resource-focused tests — pipeline deployment is tested separately. } }; } @@ -253,7 +253,7 @@ void pipelineNameNearlineToOnline() { // delete() / DependencyGuarded / specify() tests - /** Builds a 2-tier deployer with mocked CRD deployer and a pre-populated fake Database API. */ + /** Builds a 2-tier deployer with mocked custom resource deployer and a pre-populated fake Database API. */ private LogicalTableDeployer deployerWithApis(Properties props, List dbs) { return deployerWithApis(props, dbs, mockContext()); } @@ -263,48 +263,48 @@ private LogicalTableDeployer deployerWithApis(Properties props, List tierMap) { - return mockCrdDeployer; + String crName, String databaseLabel, Map tierMap) { + return mockCrDeployer; } }; } @Test - void existsDelegatesToCrdDeployer() throws SQLException { + void existsDelegatesToCrDeployer() throws SQLException { LogicalTableDeployer deployer = deployerWithApis( twoTierProps("kafka-db", "venice-db"), Arrays.asList(makeDb("kafka-db", "KAFKA"), makeDb("venice-db", "VENICE"))); - // The logical table's identity is its LogicalTable CRD; exists() reflects that CRD's presence. - when(mockCrdDeployer.exists()).thenReturn(true); + // The logical table's identity is its LogicalTable custom resource; exists() reflects that custom resource's presence. + when(mockCrDeployer.exists()).thenReturn(true); assertTrue(deployer.exists()); - verify(mockCrdDeployer).exists(); + verify(mockCrDeployer).exists(); } @Test - void existsReturnsFalseWhenCrdAbsent() throws SQLException { + void existsReturnsFalseWhenCrAbsent() throws SQLException { LogicalTableDeployer deployer = deployerWithApis( twoTierProps("kafka-db", "venice-db"), Arrays.asList(makeDb("kafka-db", "KAFKA"), makeDb("venice-db", "VENICE"))); - when(mockCrdDeployer.exists()).thenReturn(false); + when(mockCrDeployer.exists()).thenReturn(false); assertFalse(deployer.exists()); } @Test - void deleteThrowsWhenCrdDeleteFails() throws SQLException { + void deleteThrowsWhenCrDeleteFails() throws SQLException { LogicalTableDeployer deployer = deployerWithApis( twoTierProps("kafka-db", "venice-db"), Arrays.asList(makeDb("kafka-db", "KAFKA"), makeDb("venice-db", "VENICE"))); - doThrow(new SQLException("crd gone")).when(mockCrdDeployer).delete(); + doThrow(new SQLException("cr gone")).when(mockCrDeployer).delete(); SQLException ex = assertThrows(SQLException.class, deployer::delete); - assertTrue(ex.getMessage().contains("crd gone")); - verify(mockCrdDeployer).delete(); - // Tier deployers must not run when the CRD delete itself fails. + assertTrue(ex.getMessage().contains("cr gone")); + verify(mockCrDeployer).delete(); + // Tier deployers must not run when the custom resource delete itself fails. deploymentServiceMock.verify( () -> DeploymentService.deployers(any(), any()), never()); } @@ -326,7 +326,7 @@ void deleteSwallowsTierDeleteFailuresAndContinues() throws SQLException { // Must NOT throw despite the failing tier. deployer.delete(); - verify(mockCrdDeployer).delete(); + verify(mockCrDeployer).delete(); verify(failing).delete(); verify(succeeding).delete(); } @@ -407,7 +407,7 @@ void deleteKeepsSchemaEntryForTierWhoseDeleteFailed() throws SQLException { } @Test - void deleteRunsCrdDeleteBeforeTierDeletes() throws SQLException { + void deleteRunsCrDeleteBeforeTierDeletes() throws SQLException { LogicalTableDeployer deployer = deployerWithApis( twoTierProps("kafka-db", "venice-db"), Arrays.asList(makeDb("kafka-db", "KAFKA"), makeDb("venice-db", "VENICE"))); @@ -418,12 +418,12 @@ void deleteRunsCrdDeleteBeforeTierDeletes() throws SQLException { deployer.delete(); - InOrder inOrder = inOrder(mockCrdDeployer, tierDeployer); - inOrder.verify(mockCrdDeployer).delete(); + InOrder inOrder = inOrder(mockCrDeployer, tierDeployer); + inOrder.verify(mockCrDeployer).delete(); inOrder.verify(tierDeployer, atLeastOnce()).delete(); } - // CRD model construction tests + // custom resource model construction tests @Test void logicalTableSpecTierBindings() { @@ -437,7 +437,7 @@ void logicalTableSpecTierBindings() { } @Test - void crdNameIsCanonicalizedFromPath() { + void crNameIsCanonicalizedFromPath() { assertEquals("logical-mytable", K8sUtils.canonicalizeName(Arrays.asList("LOGICAL", "MyTable"))); } @@ -525,21 +525,21 @@ void buildTierMapOnlyContainsRecognizedTiers() { assertFalse(tierMap.containsKey("database")); } - // K8s-backed tests (FakeK8sApi + mock K8sContext + mock CRD deployer) + // K8s-backed tests (FakeK8sApi + mock K8sContext + mock custom resource deployer) @Test - void createDeploysLogicalTableCrd() throws Exception { + void createDeploysLogicalTableCr() throws Exception { FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).createAndReference(); + doReturn(ownerRef).when(mockCrDeployer).createAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.create(); - verify(mockCrdDeployer).createAndReference(); + verify(mockCrDeployer).createAndReference(); } @Test @@ -548,113 +548,113 @@ void createSetsDatabaseLabel() throws Exception { new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).createAndReference(); + doReturn(ownerRef).when(mockCrDeployer).createAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.create(); - // CRD content (including DATABASE_LABEL) is verified in K8sLogicalTableDeployerTest; + // custom resource content (including DATABASE_LABEL) is verified in K8sLogicalTableDeployerTest; // here we just confirm createAndReference() was called (the factory was invoked). - verify(mockCrdDeployer).createAndReference(); + verify(mockCrDeployer).createAndReference(); } @Test - void restoreDeletesCreatedCrd() throws Exception { + void restoreDeletesCreatedCr() throws Exception { FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).createAndReference(); + doReturn(ownerRef).when(mockCrDeployer).createAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.create(); deployer.restore(); - verify(mockCrdDeployer).restore(); + verify(mockCrDeployer).restore(); } @Test - void restoreAfterFailedUpdateDoesNotDeletePreExistingCrd() throws Exception { + void restoreAfterFailedUpdateDoesNotDeletePreExistingCr() throws Exception { // Regression test: restore() WITHOUT create()/update() ever running must not touch - // the CRD deployer (logicalTableDeployer is null until deployAll() runs). + // the custom resource deployer (logicalTableDeployer is null until deployAll() runs). FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); // restore() called WITHOUT create()/update() ever running deployer.restore(); - verify(mockCrdDeployer, never()).restore(); + verify(mockCrDeployer, never()).restore(); } @Test - void restoreAfterFailedCreateDeletesCrd() throws Exception { + void restoreAfterFailedCreateDeletesCr() throws Exception { // create() runs, then something downstream fails and restore() is called. - // The CRD deployer was set, so restore() must be called on it. + // The custom resource deployer was set, so restore() must be called on it. FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).createAndReference(); + doReturn(ownerRef).when(mockCrDeployer).createAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.create(); deployer.restore(); - verify(mockCrdDeployer).restore(); + verify(mockCrDeployer).restore(); } @Test - void restoreAfterUpdateThatCreatedMissingCrdDeletesIt() throws Exception { - // update() called, then restore() — the CRD deployer should have restore() called on it. + void restoreAfterUpdateThatCreatedMissingCrDeletesIt() throws Exception { + // update() called, then restore() — the custom resource deployer should have restore() called on it. FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).updateAndReference(); + doReturn(ownerRef).when(mockCrDeployer).updateAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.update(); deployer.restore(); - verify(mockCrdDeployer).restore(); + verify(mockCrDeployer).restore(); } @Test - void updateWithExistingCrdSucceeds() throws Exception { + void updateWithExistingCrSucceeds() throws Exception { FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).updateAndReference(); + doReturn(ownerRef).when(mockCrDeployer).updateAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.update(); - verify(mockCrdDeployer).updateAndReference(); + verify(mockCrDeployer).updateAndReference(); } @Test - void updateWithNoCrdCreatesNew() throws Exception { + void updateWithNoCrCreatesNew() throws Exception { FakeK8sApi dbApi = new FakeK8sApi<>(Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).updateAndReference(); + doReturn(ownerRef).when(mockCrDeployer).updateAndReference(); - LogicalTableDeployer deployer = deployerWithMockCrd( + LogicalTableDeployer deployer = deployerWithMockCr( testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi); deployer.update(); - // CRD content checks (name, spec) move to K8sLogicalTableDeployerTest - verify(mockCrdDeployer).updateAndReference(); + // custom resource content checks (name, spec) move to K8sLogicalTableDeployerTest + verify(mockCrDeployer).updateAndReference(); } @Test @@ -668,15 +668,15 @@ void createWithNearlineAndOnlineTiersAttemptsPipelineDeployment() throws Excepti makeDb("nearline-db", "NEARLINE"), makeDb("online-db", "ONLINE"))); V1OwnerReference ownerRef = new V1OwnerReference(); - doReturn(ownerRef).when(mockCrdDeployer).createAndReference(); + doReturn(ownerRef).when(mockCrDeployer).createAndReference(); - // Use a subclass that mocks the CRD deployer but does NOT suppress deployPipelineBundle, + // Use a subclass that mocks the custom resource deployer but does NOT suppress deployPipelineBundle, // so the pipeline path is exercised and fails due to the null connection in mockContext(). LogicalTableDeployer deployer = new LogicalTableDeployer(testSource(), props, mockContext(), dbApi) { @Override K8sLogicalTableDeployer createLogicalTableDeployer( - String crdName, String databaseLabel, Map tierMap) { - return mockCrdDeployer; + String crName, String databaseLabel, Map tierMap) { + return mockCrDeployer; } }; @@ -871,10 +871,10 @@ private static final class TriggerCapture { } /** - * Builds a LogicalTableDeployer that mocks the CRD deployer, suppresses pipeline deployment, + * Builds a LogicalTableDeployer that mocks the custom resource deployer, suppresses pipeline deployment, * and captures the implicit trigger via overrides on the package-private factory methods. * - *

        Pass {@code preExistingTriggers} to seed the trigger API with CRDs that already exist — + *

        Pass {@code preExistingTriggers} to seed the trigger API with custom resources that already exist — * controls whether {@code deployImplicitTrigger} takes the first-deploy (create) or the * re-deploy (update) path. */ @@ -891,8 +891,8 @@ private LogicalTableDeployer deployerWithJobTemplates( return new LogicalTableDeployer(src, props, ctx, dbApi) { @Override K8sLogicalTableDeployer createLogicalTableDeployer( - String crdName, String databaseLabel, Map tierMap) { - return mockCrdDeployer; + String crName, String databaseLabel, Map tierMap) { + return mockCrDeployer; } @Override @@ -953,7 +953,7 @@ void implicitTriggerCreatedOnCreatePathWhenOfflineTierAndMatchingJobTemplateExis Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), @@ -973,7 +973,7 @@ void implicitTriggerSkippedWhenNoOfflineTier() throws Exception { Properties props = new Properties(); props.setProperty(LogicalTier.NEARLINE.tierName(), "nearline-db"); props.setProperty(LogicalTier.ONLINE.tierName(), "online-db"); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), props, mockContext(), dbApi, jobTemplates, capture) @@ -990,7 +990,7 @@ void implicitTriggerSkippedWhenNoMatchingJobTemplate() throws Exception { // JobTemplate exists but declares a different database. List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("other-template", "some-other-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), @@ -1006,7 +1006,7 @@ void implicitTriggerPathUsesOfflineTierPhysicalPath() throws Exception { Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), @@ -1026,31 +1026,31 @@ void implicitTriggerCreatePathHasPausedOptionSet() throws Exception { Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi, jobTemplates, capture).create(); assertEquals("true", capture.trigger.options().get(Trigger.PAUSED_OPTION), - "create path must set PAUSED_OPTION=true so the CRD is created paused"); + "create path must set PAUSED_OPTION=true so the custom resource is created paused"); } @Test - void implicitTriggerUpdateOnExistingCrdOmitsPausedOptionAndCallsUpdate() throws Exception { - // Scenario: existing trigger CRD present — re-deploy must NOT set PAUSED_OPTION (paused + void implicitTriggerUpdateOnExistingCrOmitsPausedOptionAndCallsUpdate() throws Exception { + // Scenario: existing trigger custom resource present — re-deploy must NOT set PAUSED_OPTION (paused // state is preserved inside K8sTriggerDeployer.update()) and must call update(), not create(). FakeK8sApi dbApi = new FakeK8sApi<>( Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - // Seed a pre-existing trigger CRD matching the canonical implicit trigger name. + // Seed a pre-existing trigger custom resource matching the canonical implicit trigger name. List existing = new ArrayList<>(Collections.singletonList( new V1alpha1TableTrigger() .metadata(new V1ObjectMeta().name("logical-testevent-offline-trigger")))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).updateAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).updateAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), @@ -1059,31 +1059,31 @@ void implicitTriggerUpdateOnExistingCrdOmitsPausedOptionAndCallsUpdate() throws assertFalse(capture.trigger.options().containsKey(Trigger.PAUSED_OPTION), "re-deploy of existing trigger must not set PAUSED_OPTION — paused state is preserved " + "inside K8sTriggerDeployer.update()"); - assertTrue(capture.updateCalled, "update() must be invoked when the CRD already exists"); - assertFalse(capture.createCalled, "create() must not be invoked when the CRD already exists"); + assertTrue(capture.updateCalled, "update() must be invoked when the custom resource already exists"); + assertFalse(capture.createCalled, "create() must not be invoked when the custom resource already exists"); } @Test void implicitTriggerFirstCreateOrReplaceStartsPausedAndCallsCreate() throws Exception { // Scenario: CREATE OR REPLACE TABLE on a not-yet-deployed logical table (no existing trigger - // CRD). The implicit trigger must still be created PAUSED even though the DDL routes via + // custom resource). The implicit trigger must still be created PAUSED even though the DDL routes via // LogicalTableDeployer.update(). Existence check gates the decision, not the DDL verb. FakeK8sApi dbApi = new FakeK8sApi<>( Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).updateAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).updateAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), mockContext(), dbApi, jobTemplates, capture).update(); // CREATE OR REPLACE → update() assertEquals("true", capture.trigger.options().get(Trigger.PAUSED_OPTION), - "first CREATE OR REPLACE with no existing CRD must still set PAUSED_OPTION=true"); + "first CREATE OR REPLACE with no existing custom resource must still set PAUSED_OPTION=true"); assertTrue(capture.createCalled, "first-time deploy must call create() regardless of the DDL's update flag"); - assertFalse(capture.updateCalled, "update() must not be invoked when no CRD exists yet"); + assertFalse(capture.updateCalled, "update() must not be invoked when no custom resource exists yet"); } @Test @@ -1092,7 +1092,7 @@ void implicitTriggerForwardsSourceOptionsToTriggerOptions() throws Exception { Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); // CREATE TABLE WITH ("ab.cd" "customValue", "job.properties.online.name" "testevent-online") Map sourceOptions = new HashMap<>(); @@ -1116,7 +1116,7 @@ void implicitTriggerNameFollowsLogicalOfflineConvention() throws Exception { Arrays.asList(makeDb("nearline-db", "NEARLINE"), makeDb("offline-db", "OFFLINE"))); List jobTemplates = new ArrayList<>( Collections.singletonList(makeJobTemplate("retl-offline", "offline-db"))); - doReturn(new V1OwnerReference()).when(mockCrdDeployer).createAndReference(); + doReturn(new V1OwnerReference()).when(mockCrDeployer).createAndReference(); TriggerCapture capture = new TriggerCapture(); deployerWithJobTemplates(testSource(), twoTierProps("nearline-db", "offline-db"), diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableSchemaTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableSchemaTest.java index 78e6dec8..af3093bd 100644 --- a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableSchemaTest.java +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableSchemaTest.java @@ -29,7 +29,7 @@ /** * Unit tests for {@link LogicalTableSchema} label-based filtering. * - *

        Tests call {@link LogicalTableSchema#tableFromCrd} directly on a real schema + *

        Tests call {@link LogicalTableSchema#tableFromCr} directly on a real schema * instance (constructed with null K8s context) so that the production filtering * logic is tested without a live cluster. */ @@ -44,133 +44,133 @@ public void setUp() { schema = new LogicalTableSchema(new Properties(), null, DATABASE_NAME); } - // ── tableFromCrd() ─────────────────────────────────────────────────────── + // ── tableFromCr() ─────────────────────────────────────────────────────── @Test - public void tableFromCrdReturnsNullWhenMetadataIsNull() { - V1alpha1LogicalTable crd = new V1alpha1LogicalTable(); - assertNull(schema.tableFromCrd(crd)); + public void tableFromCrReturnsNullWhenMetadataIsNull() { + V1alpha1LogicalTable cr = new V1alpha1LogicalTable(); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsNullWhenSpecIsNull() { - V1alpha1LogicalTable crd = new V1alpha1LogicalTable() + public void tableFromCrReturnsNullWhenSpecIsNull() { + V1alpha1LogicalTable cr = new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name("some-table")); - assertNull(schema.tableFromCrd(crd)); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsNullWhenDatabaseLabelDoesNotMatch() { - V1alpha1LogicalTable crd = buildValidCrd("some-table", "other-database"); - assertNull(schema.tableFromCrd(crd)); + public void tableFromCrReturnsNullWhenDatabaseLabelDoesNotMatch() { + V1alpha1LogicalTable cr = buildValidCr("some-table", "other-database"); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsNullWhenDatabaseLabelIsMissing() { - V1alpha1LogicalTable crd = new V1alpha1LogicalTable() + public void tableFromCrReturnsNullWhenDatabaseLabelIsMissing() { + V1alpha1LogicalTable cr = new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name("some-table")) .spec(buildSpecWithOneTier()); - assertNull(schema.tableFromCrd(crd)); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdIsCaseInsensitiveOnLabelMatching() { - V1alpha1LogicalTable crd = buildValidCrd("some-table", "LOGICAL"); - assertNotNull(schema.tableFromCrd(crd)); + public void tableFromCrIsCaseInsensitiveOnLabelMatching() { + V1alpha1LogicalTable cr = buildValidCr("some-table", "LOGICAL"); + assertNotNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsNullWhenTiersAreNull() { - V1alpha1LogicalTable crd = new V1alpha1LogicalTable() + public void tableFromCrReturnsNullWhenTiersAreNull() { + V1alpha1LogicalTable cr = new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name("some-table") .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, DATABASE_NAME)) .spec(new V1alpha1LogicalTableSpec()); - assertNull(schema.tableFromCrd(crd)); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsNullWhenTiersAreEmpty() { + public void tableFromCrReturnsNullWhenTiersAreEmpty() { V1alpha1LogicalTableSpec spec = new V1alpha1LogicalTableSpec() .tiers(new HashMap<>()); - V1alpha1LogicalTable crd = new V1alpha1LogicalTable() + V1alpha1LogicalTable cr = new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name("some-table") .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, DATABASE_NAME)) .spec(spec); - assertNull(schema.tableFromCrd(crd)); + assertNull(schema.tableFromCr(cr)); } @Test - public void tableFromCrdReturnsLogicalTableWhenAllChecksPass() { - V1alpha1LogicalTable crd = buildValidCrd("some-table", DATABASE_NAME); - Table table = schema.tableFromCrd(crd); + public void tableFromCrReturnsLogicalTableWhenAllChecksPass() { + V1alpha1LogicalTable cr = buildValidCr("some-table", DATABASE_NAME); + Table table = schema.tableFromCr(cr); assertNotNull(table); assertTrue(table instanceof LogicalTable); } @Test - public void tableFromCrdUsesCorrectTableName() { - V1alpha1LogicalTable crd = buildValidCrd("my-table-name", DATABASE_NAME); - Table table = schema.tableFromCrd(crd); + public void tableFromCrUsesCorrectTableName() { + V1alpha1LogicalTable cr = buildValidCr("my-table-name", DATABASE_NAME); + Table table = schema.tableFromCr(cr); assertEquals("my-table-name", ((LogicalTable) table).name()); } - // ── Legacy filterCrds-style tests (now exercised via tableFromCrd) ──────── + // ── Legacy filterCrs-style tests (now exercised via tableFromCr) ──────── @Test public void tableWithMatchingLabelIsIncluded() { - V1alpha1LogicalTable crd = makeCrd("myTable", "LOGICAL"); + V1alpha1LogicalTable cr = makeCr("myTable", "LOGICAL"); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, "LOGICAL"); - Table table = s.tableFromCrd(crd); + Table table = s.tableFromCr(cr); assertNotNull(table); assertTrue(table instanceof LogicalTable); } @Test public void tableWithDifferentLabelIsExcluded() { - V1alpha1LogicalTable crd = makeCrd("otherTable", "LOGICAL-NEARLINE-OFFLINE"); + V1alpha1LogicalTable cr = makeCr("otherTable", "LOGICAL-NEARLINE-OFFLINE"); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, "LOGICAL"); - assertNull(s.tableFromCrd(crd)); + assertNull(s.tableFromCr(cr)); } @Test public void tableWithNoLabelIsExcluded() { - V1alpha1LogicalTable crd = makeCrd("unlabeled", null); + V1alpha1LogicalTable cr = makeCr("unlabeled", null); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, "LOGICAL"); - assertNull(s.tableFromCrd(crd)); + assertNull(s.tableFromCr(cr)); } @Test public void labelMatchingIsCaseInsensitive() { - V1alpha1LogicalTable crd = makeCrd("myTable", "LOGICAL"); + V1alpha1LogicalTable cr = makeCr("myTable", "LOGICAL"); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, "logical"); - assertNotNull(s.tableFromCrd(crd)); + assertNotNull(s.tableFromCr(cr)); } @Test - public void crdWithNullSpecIsSkipped() { + public void crWithNullSpecIsSkipped() { V1alpha1LogicalTable noSpec = new V1alpha1LogicalTable(); noSpec.setMetadata(new V1ObjectMeta().name("broken") .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, "LOGICAL")); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, "LOGICAL"); - assertNull(s.tableFromCrd(noSpec)); + assertNull(s.tableFromCr(noSpec)); } // ── K8s-backed tests (FakeK8sApi) ──────────────────────────────────────── @Test public void loadAllTablesViaGetTableMap() { - // Arrange: FakeK8sApi backed by a single CRD with matching label - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("logical-testevent", "testevent", DATABASE_NAME)); + // Arrange: FakeK8sApi backed by a single custom resource with matching label + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("logical-testevent", "testevent", DATABASE_NAME)); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, DATABASE_NAME, fakeApi); // Act Map tableMap = s.getTableMap(); - // Assert: the CRD spec.tableName "testevent" appears as the key, not the metadata.name + // Assert: the custom resource spec.tableName "testevent" appears as the key, not the metadata.name assertEquals(1, tableMap.size()); assertTrue(tableMap.containsKey("testevent")); } @@ -179,10 +179,10 @@ public void loadAllTablesViaGetTableMap() { public void tableNameFromSpecNotMetadataName() { // spec.tableName = "testevent", metadata.name = "logical-testevent" // The map key must be "testevent" (spec.tableName), not "logical-testevent" - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("logical-testevent", "testevent", DATABASE_NAME)); + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("logical-testevent", "testevent", DATABASE_NAME)); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, DATABASE_NAME, fakeApi); @@ -193,13 +193,13 @@ public void tableNameFromSpecNotMetadataName() { } @Test - public void loadTableMapFiltersOutCrdsWithMismatchedLabel() { - // Two CRDs: one matching, one with a different database label - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("logical-testevent", "testevent", DATABASE_NAME)); - crds.add(buildValidCrdWithTableName("other-event", "otherevent", "different-database")); + public void loadTableMapFiltersOutCrsWithMismatchedLabel() { + // Two custom resources: one matching, one with a different database label + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("logical-testevent", "testevent", DATABASE_NAME)); + crs.add(buildValidCrWithTableName("other-event", "otherevent", "different-database")); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, DATABASE_NAME, fakeApi); @@ -210,12 +210,12 @@ public void loadTableMapFiltersOutCrdsWithMismatchedLabel() { } @Test - public void loadTableMapReturnsEmptyWhenNoMatchingCrds() { - // CRD exists but belongs to a different database - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("other-event", "otherevent", "other-db")); + public void loadTableMapReturnsEmptyWhenNoMatchingCrs() { + // custom resource exists but belongs to a different database + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("other-event", "otherevent", "other-db")); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); LogicalTableSchema s = new LogicalTableSchema(new Properties(), null, DATABASE_NAME, fakeApi); @@ -226,11 +226,11 @@ public void loadTableMapReturnsEmptyWhenNoMatchingCrds() { @Test public void loadTableByNameDirectLookupViaTablesGet() { - // Arrange: FakeK8sApi with a CRD that maps "logical-testevent" → testevent - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("logical-testevent", "testevent", DATABASE_NAME)); + // Arrange: FakeK8sApi with a custom resource that maps "logical-testevent" → testevent + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("logical-testevent", "testevent", DATABASE_NAME)); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); K8sContext ctx = mock(K8sContext.class); when(ctx.namespace()).thenReturn("default"); LogicalTableSchema s = @@ -246,11 +246,11 @@ public void loadTableByNameDirectLookupViaTablesGet() { } @Test - public void loadTableByNameReturnsNullWhenCrdNotFound() { - // FakeK8sApi has no CRD named "logical-unknown" - List crds = new ArrayList<>(); + public void loadTableByNameReturnsNullWhenCrNotFound() { + // FakeK8sApi has no custom resource named "logical-unknown" + List crs = new ArrayList<>(); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); K8sContext ctx = mock(K8sContext.class); when(ctx.namespace()).thenReturn("default"); LogicalTableSchema s = @@ -273,28 +273,28 @@ public void getTableMapReturnsEmptyMapOnK8sException() { } @Test - public void tableFromCrdUsesSpecTableNameWhenPresent() { + public void tableFromCrUsesSpecTableNameWhenPresent() { // spec.tableName overrides metadata.name - V1alpha1LogicalTable crd = buildValidCrdWithTableName("compound-metadata-name", "simple", DATABASE_NAME); - Table table = schema.tableFromCrd(crd); + V1alpha1LogicalTable cr = buildValidCrWithTableName("compound-metadata-name", "simple", DATABASE_NAME); + Table table = schema.tableFromCr(cr); assertNotNull(table); assertEquals("simple", ((LogicalTable) table).name()); } @Test - public void tableFromCrdFallsBackToMetadataNameWhenTableNameNull() { + public void tableFromCrFallsBackToMetadataNameWhenTableNameNull() { // spec.tableName is null → use metadata.name V1alpha1LogicalTableSpec spec = new V1alpha1LogicalTableSpec() .putTiersItem("nearline", new V1alpha1LogicalTableSpecTiers().database("kafka-db")) .putTiersItem("online", new V1alpha1LogicalTableSpecTiers().database("venice-db")); // tableName is not set, so spec.getTableName() returns null - V1alpha1LogicalTable crd = new V1alpha1LogicalTable() + V1alpha1LogicalTable cr = new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name("logical-testevent") .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, DATABASE_NAME)) .spec(spec); - Table table = schema.tableFromCrd(crd); + Table table = schema.tableFromCr(cr); assertNotNull(table); assertEquals("logical-testevent", ((LogicalTable) table).name()); @@ -305,10 +305,10 @@ public void tierPropertyIsPassedToResolvedTierInLogicalTable() { // When TIER_PROPERTY is set, the LogicalTable should be created with that resolvedTier Properties props = new Properties(); props.setProperty(LogicalTableDriver.TIER_PROPERTY, "nearline"); - List crds = new ArrayList<>(); - crds.add(buildValidCrdWithTableName("logical-testevent", "testevent", DATABASE_NAME)); + List crs = new ArrayList<>(); + crs.add(buildValidCrWithTableName("logical-testevent", "testevent", DATABASE_NAME)); FakeK8sApi fakeApi = - new FakeK8sApi<>(crds); + new FakeK8sApi<>(crs); LogicalTableSchema s = new LogicalTableSchema(props, null, DATABASE_NAME, fakeApi); @@ -320,7 +320,7 @@ public void tierPropertyIsPassedToResolvedTierInLogicalTable() { // ── helpers ────────────────────────────────────────────────────────────── - private V1alpha1LogicalTable buildValidCrdWithTableName( + private V1alpha1LogicalTable buildValidCrWithTableName( String metadataName, String tableName, String databaseLabel) { V1alpha1LogicalTableSpec spec = new V1alpha1LogicalTableSpec() .tableName(tableName) @@ -332,7 +332,7 @@ private V1alpha1LogicalTable buildValidCrdWithTableName( .spec(spec); } - private V1alpha1LogicalTable buildValidCrd(String name, String databaseLabel) { + private V1alpha1LogicalTable buildValidCr(String name, String databaseLabel) { return new V1alpha1LogicalTable() .metadata(new V1ObjectMeta().name(name) .putLabelsItem(LogicalTableDriver.DATABASE_LABEL, databaseLabel)) @@ -344,17 +344,17 @@ private V1alpha1LogicalTableSpec buildSpecWithOneTier() { .putTiersItem("nearline", new V1alpha1LogicalTableSpecTiers().database("kafka-db")); } - private V1alpha1LogicalTable makeCrd(String name, String schemaLabel) { - V1alpha1LogicalTable crd = new V1alpha1LogicalTable(); + private V1alpha1LogicalTable makeCr(String name, String schemaLabel) { + V1alpha1LogicalTable cr = new V1alpha1LogicalTable(); V1ObjectMeta meta = new V1ObjectMeta().name(name); if (schemaLabel != null) { meta.putLabelsItem(LogicalTableDriver.DATABASE_LABEL, schemaLabel); } - crd.setMetadata(meta); + cr.setMetadata(meta); V1alpha1LogicalTableSpec spec = new V1alpha1LogicalTableSpec(); spec.putTiersItem("nearline", new V1alpha1LogicalTableSpecTiers().database("kafka-database")); spec.putTiersItem("online", new V1alpha1LogicalTableSpecTiers().database("venice")); - crd.setSpec(spec); - return crd; + cr.setSpec(spec); + return cr; } } 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 index c828765e..64102ae0 100644 --- a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableServiceIntegrationTest.java +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableServiceIntegrationTest.java @@ -26,7 +26,7 @@ /** * 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} + * pipeline custom resource/elements (querying the {@code k8s.pipelines} / {@code k8s.pipeline_element_map} * metadata tables) and the online (Venice) tier's resolved row type. */ @Tag("integration") @@ -103,7 +103,7 @@ void onlineLogicalTableLifecycle() throws SQLException { 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 + // The deployment is also verified structurally via the pipeline custom resource and its elements: the // nearline KafkaTopic physical table plus the identity FlinkSessionJob. assertThat(pipelineNames()).contains(pipeline); List elements = pipelineElements(pipeline); diff --git a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableTest.java b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableTest.java index ea100489..5dce725e 100644 --- a/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableTest.java +++ b/hoptimator-logical/src/test/java/com/linkedin/hoptimator/logical/LogicalTableTest.java @@ -128,13 +128,13 @@ public void unknownTypeReturnedWhenTierBindingHasNullDatabaseName() { } @Test - public void unknownTypeReturnedWhenDbCrdHasNullSpec() { - // FakeK8sApi returns a Database CRD with null spec → resolveRowType returns null → unknown type - V1alpha1Database dbCrd = new V1alpha1Database() + public void unknownTypeReturnedWhenDbCrHasNullSpec() { + // FakeK8sApi returns a Database custom resource with null spec → resolveRowType returns null → unknown type + V1alpha1Database dbCr = new V1alpha1Database() .metadata(new V1ObjectMeta().name("nearline-db")); // spec is null List databases = new ArrayList<>(); - databases.add(dbCrd); + databases.add(dbCr); FakeK8sApi fakeDbApi = new FakeK8sApi<>(databases); Map tiers = new HashMap<>(); @@ -147,15 +147,15 @@ public void unknownTypeReturnedWhenDbCrdHasNullSpec() { @Test public void unknownTypeReturnedWhenJdbcUrlNotFound() { - // FakeK8sApi returns a Database CRD with a bad URL → DriverManager.getConnection fails + // FakeK8sApi returns a Database custom resource with a bad URL → DriverManager.getConnection fails // → catch block → returns null → unknown type - V1alpha1Database dbCrd = new V1alpha1Database() + V1alpha1Database dbCr = new V1alpha1Database() .metadata(new V1ObjectMeta().name("nearline-db")) .spec(new V1alpha1DatabaseSpec() .url("jdbc:nonexistent://host") .schema("PROFILE")); List databases = new ArrayList<>(); - databases.add(dbCrd); + databases.add(dbCr); FakeK8sApi fakeDbApi = new FakeK8sApi<>(databases); @@ -169,14 +169,14 @@ public void unknownTypeReturnedWhenJdbcUrlNotFound() { @Test public void unknownTypeReturnedWhenSchemaNotFoundInTier() { - // Database CRD with valid URL (demodb) but schema name that doesn't exist - V1alpha1Database dbCrd = new V1alpha1Database() + // Database custom resource with valid URL (demodb) but schema name that doesn't exist + V1alpha1Database dbCr = new V1alpha1Database() .metadata(new V1ObjectMeta().name("nearline-db")) .spec(new V1alpha1DatabaseSpec() .url("jdbc:demodb://names=PROFILE") .schema("NONEXISTENT_SCHEMA")); List databases = new ArrayList<>(); - databases.add(dbCrd); + databases.add(dbCr); FakeK8sApi fakeDbApi = new FakeK8sApi<>(databases); @@ -191,14 +191,14 @@ public void unknownTypeReturnedWhenSchemaNotFoundInTier() { @Test public void unknownTypeReturnedWhenTableNotFoundInTierSchema() { - // Database CRD with valid URL (demodb) and valid schema but table name doesn't exist - V1alpha1Database dbCrd = new V1alpha1Database() + // Database custom resource with valid URL (demodb) and valid schema but table name doesn't exist + V1alpha1Database dbCr = new V1alpha1Database() .metadata(new V1ObjectMeta().name("nearline-db")) .spec(new V1alpha1DatabaseSpec() .url("jdbc:demodb://names=PROFILE") .schema("PROFILE")); List databases = new ArrayList<>(); - databases.add(dbCrd); + databases.add(dbCr); FakeK8sApi fakeDbApi = new FakeK8sApi<>(databases); @@ -213,14 +213,14 @@ public void unknownTypeReturnedWhenTableNotFoundInTierSchema() { @Test public void rowTypeResolvedFromTierWhenTableExistsInDemodb() { - // Full happy path: Database CRD → demodb PROFILE schema → MEMBERS table → row type resolved - V1alpha1Database dbCrd = new V1alpha1Database() + // Full happy path: Database custom resource → demodb PROFILE schema → MEMBERS table → row type resolved + V1alpha1Database dbCr = new V1alpha1Database() .metadata(new V1ObjectMeta().name("nearline-db")) .spec(new V1alpha1DatabaseSpec() .url("jdbc:demodb://names=PROFILE") .schema("PROFILE")); List databases = new ArrayList<>(); - databases.add(dbCrd); + databases.add(dbCr); FakeK8sApi fakeDbApi = new FakeK8sApi<>(databases); diff --git a/hoptimator-mysql/src/main/java/com/linkedin/hoptimator/mysql/MySqlDeployerProvider.java b/hoptimator-mysql/src/main/java/com/linkedin/hoptimator/mysql/MySqlDeployerProvider.java index 31bb31b7..c06d813d 100644 --- a/hoptimator-mysql/src/main/java/com/linkedin/hoptimator/mysql/MySqlDeployerProvider.java +++ b/hoptimator-mysql/src/main/java/com/linkedin/hoptimator/mysql/MySqlDeployerProvider.java @@ -18,7 +18,7 @@ *

        Detection uses {@link DeploymentContext#databaseProperties} to resolve the source's * {@code Database} config and checks whether its connection URL starts with * {@code jdbc:mysql-hoptimator://}. This is Calcite-free: it works identically whether the config - * comes from the Calcite catalog (SQL path) or from a {@code Database} CRD (direct path). + * comes from the Calcite catalog (SQL path) or from a {@code Database} custom resource (direct path). */ public class MySqlDeployerProvider implements DeployerProvider { diff --git a/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/LogicalSchemaMarker.java b/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/LogicalSchemaMarker.java index d369aad5..942cfb3c 100644 --- a/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/LogicalSchemaMarker.java +++ b/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/LogicalSchemaMarker.java @@ -2,11 +2,11 @@ /** * Marker interface implemented by Calcite {@code Schema} instances that surface Hoptimator - * LogicalTable-style CRDs — i.e. schemas whose tables aren't physical resources but logical + * LogicalTable-style custom resources — i.e. schemas whose tables aren't physical resources but logical * declarations bound to one or more tiers. * *

        Why this exists. {@link HoptimatorJdbcSchema} is the outer JDBC adapter that - * fronts every {@code Database} CRD in the user's connection. When the underlying driver + * fronts every {@code Database} custom resource in the user's connection. When the underlying driver * surfaces logical tables (e.g. {@code jdbc:logical://...}), the outer adapter needs a way * to tell — without baking driver-specific URL prefixes or class names into its own logic. * 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 d1c5b301..85950082 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 @@ -198,7 +198,7 @@ public void restore() { } protected Pair getKeyPayloadSchema() throws SQLException { - Map keyOptions = resolveKeyOptions(); + Map keyOptions = ConnectionService.configure(source, context); // 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); @@ -212,15 +212,6 @@ protected Pair getKeyPayloadSchema() throws SQLException { keyOptions); } - /** - * Resolves connector-supplied options (e.g. the {@code keys} option that drives the key/payload - * split) for this store. Extracted so unit tests can supply options directly instead of resolving - * them through the live {@link ConnectionService} connector chain, which would reach out to K8s. - */ - protected Map resolveKeyOptions() throws SQLException { - return ConnectionService.configure(source, context); - } - private boolean checkStoreExists(ControllerClient controllerClient) { try { StoreResponse response = controllerClient.getStore(source.table()); 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 a9be7e24..bf564f6a 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 @@ -5,6 +5,7 @@ import com.linkedin.hoptimator.jdbc.CalciteDeploymentContext; import com.linkedin.hoptimator.jdbc.DirectDeploymentContext; import com.linkedin.hoptimator.jdbc.HoptimatorConnection; +import com.linkedin.hoptimator.util.ConnectionService; import com.linkedin.venice.client.schema.StoreSchemaFetcher; import com.linkedin.venice.controllerapi.ControllerClient; import com.linkedin.venice.controllerapi.ControllerResponse; @@ -28,13 +29,13 @@ 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.sql.SQLNonTransientException; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Properties; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -42,6 +43,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -69,6 +71,9 @@ class VeniceDeployerTest { @Mock private StoreSchemaFetcher mockStoreSchemaFetcher; + @Mock + private MockedStatic connectionService; + private Properties properties; @BeforeEach @@ -335,15 +340,12 @@ public void testGetKeyPayloadSchemaPrefersProvidedAvroLosslessly() throws Except + "{\"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(); - } - }; + + // Stub the K8s-reaching option resolution instead of overriding a + // deployer method; the provided-Avro path under test does not depend on key options. + connectionService.when(() -> ConnectionService.configure(any(), any())).thenReturn(Collections.emptyMap()); + VeniceDeployer deployer = + new VeniceDeployer(source, properties, new DirectDeploymentContext(properties, null, provided)); Pair keyPayload = deployer.getKeyPayloadSchema(); @@ -378,16 +380,12 @@ public RelDataType getRowType(RelDataTypeFactory typeFactory) { CalciteConnection calciteConnection = mock(CalciteConnection.class); when(mockConnection.calciteConnection()).thenReturn(calciteConnection); when(calciteConnection.getRootSchema()).thenReturn(rootSchema); + + // Stub the K8s-reaching option resolution instead of overriding a + // deployer method; with no keys, the whole row type is treated as the payload. + connectionService.when(() -> ConnectionService.configure(any(), any())).thenReturn(Collections.emptyMap()); VeniceDeployer deployer = - new VeniceDeployer(source, properties, new CalciteDeploymentContext(mockConnection)) { - @Override - protected Map resolveKeyOptions() { - // Unit-test seam: no connector on this classpath supplies a "keys" option, and resolving - // through the live ConnectionService would reach out to K8s. Return none so the whole - // row type is treated as the payload. - return Collections.emptyMap(); - } - }; + new VeniceDeployer(source, properties, new CalciteDeploymentContext(mockConnection)); Pair keyPayload = deployer.getKeyPayloadSchema();