diff --git a/‎.github/workflows/integration-tests.yml‎
Lines changed: 12 additions & 0 deletions b/‎.github/workflows/integration-tests.yml‎
Lines changed: 12 additions & 0 deletions
diff --git a/‎Makefile‎
Lines changed: 6 additions & 2 deletions b/‎Makefile‎
Lines changed: 6 additions & 2 deletions
diff --git a/‎build.gradle‎
Lines changed: 58 additions & 1 deletion b/‎build.gradle‎
Lines changed: 58 additions & 1 deletion
diff --git a/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/JobTest.java‎
Lines changed: 66 additions & 0 deletions b/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/JobTest.java‎
Lines changed: 66 additions & 0 deletions
diff --git a/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/MaterializedViewTest.java‎
Lines changed: 36 additions & 0 deletions b/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/MaterializedViewTest.java‎
Lines changed: 36 additions & 0 deletions
diff --git a/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/PipelineTest.java‎
Lines changed: 28 additions & 0 deletions b/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/PipelineTest.java‎
Lines changed: 28 additions & 0 deletions
diff --git a/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/SinkTest.java‎
Lines changed: 27 additions & 0 deletions b/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/SinkTest.java‎
Lines changed: 27 additions & 0 deletions
diff --git a/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/SourceTest.java‎
Lines changed: 84 additions & 0 deletions b/‎hoptimator-api/src/test/java/com/linkedin/hoptimator/SourceTest.java‎
Lines changed: 84 additions & 0 deletions
@@ -8,6 +8,7 @@ on:
 
 permissions:
   contents: read
+  pull-requests: write
 
 jobs:
   build:
@@ -23,6 +24,17 @@ jobs:
         distribution: 'temurin'
     - name: Build
       run: make build
+    - name: Jacoco Report to PR
+      id: jacoco
+      uses: madrapps/jacoco-report@v1.7.2
+      with:
+        paths: ${{ github.workspace }}/build/reports/jacoco/aggregate/jacocoAggregateReport.xml
+        token: ${{ secrets.GITHUB_TOKEN }}
+        title: Code Coverage
+        min-coverage-overall: 40
+        min-coverage-changed-files: 60
+        pass-emoji: ':green_circle:'
+        update-comment: true
     - name: Create K8s Kind Cluster
       uses: helm/kind-action@v1.4.0
       with:
 
@@ -5,8 +5,12 @@ install:
 test:
 	./gradlew test -x spotbugsMain -x spotbugsTest -x spotbugsTestFixtures
 
+coverage:
+	./gradlew jacocoAggregateReport -x spotbugsMain -x spotbugsTest -x spotbugsTestFixtures
+	@echo "Aggregate report: file://$(shell pwd)/build/reports/jacoco/aggregate/index.html"
+
 build:
-	./gradlew build shadowJar
+	./gradlew build jacocoAggregateReport shadowJar
 	docker build . -t hoptimator
 	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-runner -t hoptimator-flink-runner
 	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-operator -t hoptimator-flink-operator
@@ -149,4 +153,4 @@ run-zeppelin: build-zeppelin
 	  --name hoptimator-zeppelin \
 	  hoptimator-zeppelin
 
-.PHONY: install test build bounce clean quickstart deploy-config undeploy-config deploy undeploy deploy-demo undeploy-demo deploy-flink undeploy-flink deploy-kafka undeploy-kafka deploy-mysql undeploy-mysql deploy-venice undeploy-venice build-zeppelin run-zeppelin integration-tests integration-tests-kind deploy-dev-environment undeploy-dev-environment generate-models release
+.PHONY: install test coverage build bounce clean quickstart deploy-config undeploy-config deploy undeploy deploy-demo undeploy-demo deploy-flink undeploy-flink deploy-kafka undeploy-kafka deploy-mysql undeploy-mysql deploy-venice undeploy-venice build-zeppelin run-zeppelin integration-tests integration-tests-kind deploy-dev-environment undeploy-dev-environment generate-models release
@@ -2,6 +2,57 @@ plugins {
   id("com.github.spotbugs") version "6.1.10"
 }
 
+apply plugin: 'jacoco'
+
+// All JaCoCo exclusions for the aggregate report, centralised here.
+// Per-module jacocoTestReport exclusions don't reliably propagate to the aggregate
+// task because of Gradle's configuration resolution order, so all exclusions are
+// declared once and applied when building filtered FileTrees below.
+ext.jacocoAggregateExcludes = [
+  // ── hoptimator-models / hoptimator-k8s ──────────────────────────────────────
+  '**/models/**',           // auto-generated K8s CRD model classes
+
+  // ── hoptimator-operator ──────────────────────────────────────────────────────
+  '**/PipelineOperatorApp.class',                  // operator main entry point
+
+  // ── hoptimator-flink-runner ──────────────────────────────────────────────────
+  '**/FlinkRunner.class',                          // CLI entry point; requires Flink runtime
+
+  // ── hoptimator-cli ───────────────────────────────────────────────────────────
+  'sqlline/**',                                    // third-party library bundled via application
+
+  // ── hoptimator-jdbc ──────────────────────────────────────────────────────────
+  '**/ddl/**',                                     // auto-generated JavaCC SQL parser
+]
+
+task jacocoAggregateReport(type: JacocoReport) {
+  group = 'verification'
+  description = 'Generates an aggregate JaCoCo coverage report across all subprojects.'
+  reports {
+    html.required = true
+    xml.required = true
+    html.outputLocation = layout.buildDirectory.dir('reports/jacoco/aggregate')
+    xml.outputLocation = layout.buildDirectory.file('reports/jacoco/aggregate/jacocoAggregateReport.xml')
+  }
+}
+
+subprojects {
+  plugins.withType(JacocoPlugin).configureEach {
+    rootProject.jacocoAggregateReport.dependsOn(test)
+    rootProject.jacocoAggregateReport.executionData(test)
+    rootProject.jacocoAggregateReport.sourceDirectories.from(sourceSets.main.allSource.srcDirs)
+    // Add a filtered FileTree for each classes directory. Using fileTree directly (rather than
+    // jacocoTestReport.classDirectories) ensures the exclusions are applied at resolution time,
+    // bypassing Gradle's configuration-resolution ordering issue with per-module setFrom().
+    def excludes = rootProject.ext.jacocoAggregateExcludes
+    sourceSets.main.output.classesDirs.each { classesDir ->
+      rootProject.jacocoAggregateReport.classDirectories.from(
+        fileTree(dir: classesDir, excludes: excludes)
+      )
+    }
+  }
+}
+
 subprojects {
   apply plugin: 'checkstyle'
   apply plugin: 'com.github.spotbugs'
@@ -18,12 +69,18 @@ subprojects {
   spotbugs {
     ignoreFailures = false
     showProgress = true
-    reportsDir = file("$rootProject.layout.buildDirectory/spotbugs")
+    reportsDir = file("${rootDir}/build/spotbugs")
     includeFilter = file("$rootDir/config/spotbugs/include.xml")
     excludeFilter = file("$rootDir/config/spotbugs/exclude.xml")
   }
 
   plugins.withType(JavaPlugin) {
+    project.apply plugin: 'jacoco'
+
+    jacoco {
+      toolVersion = "0.8.11"
+    }
+
     dependencies {
       testImplementation libs.assertj
       testImplementation libs.junit.jupiter.api
 
@@ -0,0 +1,66 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+
+class JobTest {
+
+  @Test
+  void testNameAndAccessors() {
+    Sink sink = new Sink("db", List.of("s", "t"), Collections.emptyMap());
+    Source src = new Source("db", List.of("s", "src"), Collections.emptyMap());
+    Set<Source> sources = Set.of(src);
+    Map<String, ThrowingFunction<SqlDialect, String>> evals = Map.of(
+        "sql", d -> "INSERT INTO t SELECT * FROM src",
+        "query", d -> "SELECT * FROM src",
+        "fieldMap", d -> "{}"
+    );
+
+    Job job = new Job("myJob", sources, sink, evals);
+
+    assertEquals("myJob", job.name());
+    assertEquals(sources, job.sources());
+    assertEquals(sink, job.sink());
+  }
+
+  @Test
+  void testSqlReturnsFunction() throws Exception {
+    Sink sink = new Sink("db", List.of("t"), Collections.emptyMap());
+    Map<String, ThrowingFunction<SqlDialect, String>> evals = Map.of(
+        "sql", d -> "my-sql",
+        "query", d -> "my-query",
+        "fieldMap", d -> "my-fields"
+    );
+    Job job = new Job("j", Collections.emptySet(), sink, evals);
+
+    assertEquals("my-sql", job.sql().apply(SqlDialect.ANSI));
+    assertEquals("my-query", job.query().apply(SqlDialect.FLINK));
+    assertEquals("my-fields", job.fieldMap().apply(SqlDialect.ANSI));
+  }
+
+  @Test
+  void testEvalThrowsForUnknownKey() {
+    Sink sink = new Sink("db", List.of("t"), Collections.emptyMap());
+    Map<String, ThrowingFunction<SqlDialect, String>> evals = Map.of(
+        "sql", d -> "s"
+    );
+    Job job = new Job("j", Collections.emptySet(), sink, evals);
+
+    assertThrows(IllegalArgumentException.class, job::query);
+  }
+
+  @Test
+  void testToString() {
+    Sink sink = new Sink("db", List.of("schema", "table"), Collections.emptyMap());
+    Job job = new Job("j", Collections.emptySet(), sink, Collections.emptyMap());
+    assertEquals("Job[schema.table]", job.toString());
+  }
+}
@@ -0,0 +1,36 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+
+class MaterializedViewTest {
+
+  @Test
+  void testAccessors() {
+    Sink sink = new Sink("db", List.of("sink"), Collections.emptyMap());
+    Job job = new Job("j", Collections.emptySet(), sink, Collections.emptyMap());
+    Pipeline pipeline = new Pipeline(Collections.emptyList(), sink, job);
+    ThrowingFunction<SqlDialect, String> pSql = d -> "INSERT INTO sink SELECT * FROM src";
+
+    MaterializedView mv = new MaterializedView("myDb", List.of("s", "v"), "SELECT 1", pSql, pipeline);
+
+    assertEquals("myDb", mv.database());
+    assertEquals(pipeline, mv.pipeline());
+    assertNotNull(mv.pipelineSql());
+    assertEquals("SELECT 1", mv.viewSql());
+    assertEquals("v", mv.table());
+    assertEquals("s", mv.schema());
+  }
+
+  @Test
+  void testToString() {
+    MaterializedView mv = new MaterializedView("db", List.of("a", "b"), "SELECT 1", d -> "", null);
+    assertEquals("MaterializedView[a.b]", mv.toString());
+  }
+}
@@ -0,0 +1,28 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+
+class PipelineTest {
+
+  @Test
+  void testAccessors() {
+    Source source = new Source("db", List.of("src"), Collections.emptyMap());
+    Sink sink = new Sink("db", List.of("sink"), Collections.emptyMap());
+    Job job = new Job("j", Collections.emptySet(), sink, Collections.emptyMap());
+    List<Source> sources = List.of(source);
+
+    Pipeline pipeline = new Pipeline(sources, sink, job);
+
+    assertEquals(sources, List.copyOf(pipeline.sources()));
+    assertEquals(sink, pipeline.sink());
+    assertEquals(job, pipeline.job());
+    assertNotNull(pipeline.sources());
+  }
+}
@@ -0,0 +1,27 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class SinkTest {
+
+  @Test
+  void testToString() {
+    Sink sink = new Sink("db", List.of("schema", "table"), Collections.emptyMap());
+    assertEquals("Sink[schema.table]", sink.toString());
+  }
+
+  @Test
+  void testInheritsSourceBehavior() {
+    Sink sink = new Sink("myDb", List.of("cat", "sch", "tbl"), Collections.emptyMap());
+    assertEquals("tbl", sink.table());
+    assertEquals("sch", sink.schema());
+    assertEquals("cat", sink.catalog());
+    assertEquals("myDb", sink.database());
+  }
+}
@@ -0,0 +1,84 @@
+package com.linkedin.hoptimator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+
+class SourceTest {
+
+  @Test
+  void testTableReturnsLastPathElement() {
+    Source source = new Source("db", List.of("catalog", "schema", "myTable"), Collections.emptyMap());
+    assertEquals("myTable", source.table());
+  }
+
+  @Test
+  void testSchemaReturnsPenultimateElement() {
+    Source source = new Source("db", List.of("catalog", "schema", "myTable"), Collections.emptyMap());
+    assertEquals("schema", source.schema());
+  }
+
+  @Test
+  void testCatalogReturnsThirdFromEnd() {
+    Source source = new Source("db", List.of("catalog", "schema", "myTable"), Collections.emptyMap());
+    assertEquals("catalog", source.catalog());
+  }
+
+  @Test
+  void testCatalogReturnsNullForTwoElementPath() {
+    Source source = new Source("db", List.of("schema", "myTable"), Collections.emptyMap());
+    assertNull(source.catalog());
+  }
+
+  @Test
+  void testDatabase() {
+    Source source = new Source("myDb", List.of("t"), Collections.emptyMap());
+    assertEquals("myDb", source.database());
+  }
+
+  @Test
+  void testOptions() {
+    Map<String, String> opts = Map.of("key", "value");
+    Source source = new Source("db", List.of("t"), opts);
+    assertEquals(opts, source.options());
+  }
+
+  @Test
+  void testPath() {
+    List<String> path = List.of("a", "b", "c");
+    Source source = new Source("db", path, Collections.emptyMap());
+    assertEquals(path, source.path());
+  }
+
+  @Test
+  void testPathString() {
+    Source source = new Source("db", List.of("a", "b", "c"), Collections.emptyMap());
+    assertEquals("a.b.c", source.pathString());
+  }
+
+  @Test
+  void testToString() {
+    Source source = new Source("db", List.of("a", "b"), Collections.emptyMap());
+    assertEquals("Source[a.b]", source.toString());
+  }
+
+  @Test
+  void testSchemaReturnsValueForExactlyTwoElementPath() {
+    Source source = new Source("db", List.of("mySchema", "myTable"), Collections.emptyMap());
+    assertEquals("mySchema", source.schema(),
+        "schema() must return the second-to-last element when path has exactly 2 elements");
+  }
+
+  @Test
+  void testSchemaReturnsNullForExactlyOneElementPath() {
+    Source source = new Source("db", List.of("onlyTable"), Collections.emptyMap());
+    assertNull(source.schema(),
+        "schema() must return null when path has only 1 element");
+  }
+}