Apply scalafmt & add py entry point

.github/workflows/ci-jvm.yml

@@ -0,0 +1,26 @@
+name: jvm-server
+
+on: push
+
+jobs:
+  scalatest:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up JDK 11 for x64
+        uses: actions/setup-java@v4
+        with:
+          java-version: '11'
+          distribution: 'temurin'
+          architecture: x64
+
+      - name: Check codestyle
+        run: |
+          mvn --no-transfer-progress spotless:check
+        working-directory: tsumugi-server
+
+      - name: Run tests
+        run: |
+          mvn --no-transfer-progress test
+        working-directory: tsumugi-server
+

tsumugi-python/tsumugi/verification.py

@@ -2,6 +2,7 @@
 from dataclasses import asdict, dataclass
 
 import pandas as pd
+from py4j.java_gateway import JavaObject
 from pyspark.sql import DataFrame, SQLContext
 from pyspark.sql import functions as F
 from pyspark.sql.connect.client import SparkConnectClient
@@ -286,15 +287,10 @@ def run(self) -> VerificationResult:
         if is_classic:
             jvm = spark._jvm
             jdf = self._data._jdf
-            deequ_jvm_suite = jvm.io.mrpowers.tsumugi.DeequSuiteBuilder(
-                jdf,
-                pb_suite,
-            )
-            result_jdf = jvm.io.mrpowers.tsumugi.DeeqUtils.runAndCollectResults(
-                deequ_jvm_suite,
-                spark._jsparkSession,
-                self._compute_row_results,
-                jdf,
+            result_jdf: JavaObject = (
+                jvm.io.mrpowers.tsumugi.DeequSuiteBuilder.entryPoint(
+                    jdf, pb_suite.SerializeToString()
+                )
             )
             return VerificationResult(
                 DataFrame(result_jdf, SQLContext(spark.sparkContext))

tsumugi-server/.scalafmt.conf

@@ -1,4 +1,4 @@
-version = 3.8.2
+version = 3.7.3
 
 align.preset = some
 runner.dialect = scala213

tsumugi-server/src/main/scala/io/mrpowers/tsumugi/DeequUtils.scala

@@ -94,8 +94,10 @@ object DeequUtils {
       sparkSession: SparkSession,
       dataFrame: DataFrame
   ): DataFrame = {
-    val metrics = VerificationResult.successMetricsAsDataFrame(sparkSession = sparkSession, verificationResult = deequSuite)
-    val checks = VerificationResult.checkResultsAsDataFrame(sparkSession = sparkSession, verificationResult = deequSuite)
+    val metrics =
+      VerificationResult.successMetricsAsDataFrame(sparkSession = sparkSession, verificationResult = deequSuite)
+    val checks =
+      VerificationResult.checkResultsAsDataFrame(sparkSession = sparkSession, verificationResult = deequSuite)
     val checkResults = checkResultAsDataFramePatched(deequSuite = deequSuite, sparkSession = sparkSession)
 
     val oneRowDf =
@@ -111,7 +113,7 @@ object DeequUtils {
         withColumnFrom(df, from, ARRAY_COL, Option(alias))
       }
     if (returnRows) {
-     val rowResults = VerificationResult.rowLevelResultsAsDataFrame(
+      val rowResults = VerificationResult.rowLevelResultsAsDataFrame(
         sparkSession = sparkSession,
         verificationResult = deequSuite,
         data = dataFrame

tsumugi-server/src/main/scala/org/apache/spark/sql/tsumugi/DeequConnectPlugin.scala

@@ -18,21 +18,7 @@ class DeequConnectPlugin extends RelationPlugin {
       val spark = planner.sessionHolder.session
       val protoPlan = org.apache.spark.connect.proto.Plan.parseFrom(protoSuite.getData.toByteArray)
       val data = Dataset.ofRows(spark, planner.transformRelation(protoPlan.getRoot))
-      val verificationSuiteBuilder = DeequSuiteBuilder
-        .protoToVerificationSuite(
-          data,
-          protoSuite
-        )
-
-      // TODO: pass returnRows here
-      val resultDf =
-        DeequUtils.runAndCollectResults(
-          verificationSuiteBuilder.get,
-          spark,
-          returnRows = protoSuite.getComputeRowLevelResults,
-          dataFrame = data
-        )
-      Some(resultDf.logicalPlan)
+      Some(DeequSuiteBuilder.entryPoint(data, protoSuite).logicalPlan)
     } else {
       None
     }

tsumugi-server/src/test/scala/io/mrpowers/tsumugi/DeequSuiteBuilderTest.scala

@@ -181,7 +181,8 @@ class DeequSuiteBuilderTest extends ConfTest {
                     .build()
                 )
             )
-        ).build()
+        )
+        .build()
     )
 
     val metric = VerificationSuite()
@@ -202,7 +203,7 @@ class DeequSuiteBuilderTest extends ConfTest {
       5
     )
     metric._1 match {
-      case analyzers.Histogram(_, _, _, _, _, analyzers.Histogram.Sum(aggColumn))  =>
+      case analyzers.Histogram(_, _, _, _, _, analyzers.Histogram.Sum(aggColumn)) =>
         assert(aggColumn == "numViews")
         assert(metric._2.value.get == expectedDistribution)
       case _ => fail(s"Expected ${analyzers.Histogram} with ${analyzers.Histogram.Sum} aggregate functions")
@@ -229,7 +230,8 @@ class DeequSuiteBuilderTest extends ConfTest {
                     .build()
                 )
             )
-        ).build()
+        )
+        .build()
     )
 
     val metric = VerificationSuite()
@@ -250,7 +252,8 @@ class DeequSuiteBuilderTest extends ConfTest {
     )
 
     metric._1 match {
-      case analyzers.Histogram(_, _, _, _, _, analyzers.Histogram.Count)  => assert(metric._2.value.get == expectedDistribution)
+      case analyzers.Histogram(_, _, _, _, _, analyzers.Histogram.Count) =>
+        assert(metric._2.value.get == expectedDistribution)
       case _ => fail(s"Expected ${analyzers.Histogram} with ${analyzers.Histogram.Count} aggregate functions")
     }
   }