Apply scalafmt changes

Author: EnricoMi

pom.xml

@@ -293,6 +293,29 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>com.diffplug.spotless</groupId>
+        <artifactId>spotless-maven-plugin</artifactId>
+        <version>2.30.0</version>
+        <configuration>
+          <scala>
+            <scalafmt>
+              <version>3.7.17</version>
+              <file>${project.basedir}/.scalafmt.conf</file>
+            </scalafmt>
+          </scala>
+        </configuration>
+        <executions>
+          <execution>
+            <!-- Runs in compile phase to fail fast in case of formatting issues.-->
+            <id>spotless-check</id>
+            <phase>compile</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 

src/main/scala/uk/co/gresearch/spark/dgraph/connector/ClusterState.scala

@@ -23,10 +23,12 @@ import java.util.UUID
 import scala.jdk.CollectionConverters._
 import scala.util.{Failure, Success, Try}
 
-case class ClusterState(groupMembers: Map[String, Set[Target]],
-                        groupPredicates: Map[String, Set[String]],
-                        maxUid: Option[UnsignedLong],
-                        cid: UUID)
+case class ClusterState(
+    groupMembers: Map[String, Set[Target]],
+    groupPredicates: Map[String, Set[String]],
+    maxUid: Option[UnsignedLong],
+    cid: UUID
+)
 
 object ClusterState extends Logging {
 
@@ -35,30 +37,38 @@ object ClusterState extends Logging {
     val groups = root.getAsJsonObject("groups")
     val groupMap =
       groups
-        .entrySet().asScala
+        .entrySet()
+        .asScala
         .map(e => e.getKey -> e.getValue.getAsJsonObject)
         .toMap
 
-    val groupMembers = groupMap.map { case (key, group) => key -> getMembersFromGroup(group) }
+    val groupMembers = groupMap
+      .map { case (key, group) => key -> getMembersFromGroup(group) }
       .map { case (key, members) => key -> members.map(Target).map(t => t.withPort(t.port + 2000)) }
     val groupPredicates = groupMap.map { case (key, group) => key -> getPredicatesFromGroup(group) }
     val cid = UUID.fromString(root.getAsJsonPrimitive("cid").getAsString)
 
     val maxUid = (if (root.has("maxUID")) {
-      getUnsignedLongFromJson(root.getAsJsonPrimitive("maxUID"))
-    } else if (root.has("maxLeaseId")) {
-      getUnsignedLongFromJson(root.getAsJsonPrimitive("maxLeaseId"))
-    } else {
-      Failure(new IllegalArgumentException("Cluster state does not contain maxLeaseId or maxUID, this disables uid range partitioning"))
-    }) match {
+                    getUnsignedLongFromJson(root.getAsJsonPrimitive("maxUID"))
+                  } else if (root.has("maxLeaseId")) {
+                    getUnsignedLongFromJson(root.getAsJsonPrimitive("maxLeaseId"))
+                  } else {
+                    Failure(
+                      new IllegalArgumentException(
+                        "Cluster state does not contain maxLeaseId or maxUID, this disables uid range partitioning"
+                      )
+                    )
+                  }) match {
       case Success(value) => Some(value)
       case Failure(exception) =>
         log.error("Failed to retrieve maxUID from cluster state", exception)
         None
     }
 
     if (maxUid.exists(_.compareTo(UnsignedLong.ZERO) < 0)) {
-      log.error(s"Cluster state indicates negative maxLeaseId or maxUID, this disables uid range partitioning: ${maxUid.get}")
+      log.error(
+        s"Cluster state indicates negative maxLeaseId or maxUID, this disables uid range partitioning: ${maxUid.get}"
+      )
     }
     val positiveMaxUID = maxUid.filter(_.compareTo(UnsignedLong.ZERO) >= 0)
 

src/main/scala/uk/co/gresearch/spark/dgraph/connector/ClusterStateProvider.scala

@@ -26,10 +26,14 @@ trait ClusterStateProvider extends Logging {
     val clusterStates = targets.flatMap(target => getClusterState(target))
     val cids = clusterStates.map(_.cid).toSet
     if (cids.size > 1)
-      throw new RuntimeException(s"Retrieved multiple cluster ids from " +
-        s"Dgraph alphas (${targets.map(_.target).mkString(", ")}): ${cids.mkString(", ")}")
+      throw new RuntimeException(
+        s"Retrieved multiple cluster ids from " +
+          s"Dgraph alphas (${targets.map(_.target).mkString(", ")}): ${cids.mkString(", ")}"
+      )
     val clusterState = clusterStates.headOption.getOrElse(
-      throw new RuntimeException(s"Could not retrieve cluster state from Dgraph alphas (${targets.map(_.target).mkString(", ")})")
+      throw new RuntimeException(
+        s"Could not retrieve cluster state from Dgraph alphas (${targets.map(_.target).mkString(", ")})"
+      )
     )
 
     // filter out reserved predicates as configured
@@ -42,16 +46,18 @@ trait ClusterStateProvider extends Logging {
   }
 
   def getClusterState(target: Target): Option[ClusterState] = {
-    val url = s"http://${target.withPort(target.port-1000).target}/state"
+    val url = s"http://${target.withPort(target.port - 1000).target}/state"
     try {
       val startTs = Clock.systemUTC().instant().toEpochMilli
       val request = requests.get(url)
       val endTs = Clock.systemUTC().instant().toEpochMilli
       val json = Json(request.text())
 
-      log.info(s"retrieved cluster state from ${target.target} " +
-        s"with ${json.string.getBytes.length} bytes " +
-        s"in ${(endTs - startTs) / 1000.0}s")
+      log.info(
+        s"retrieved cluster state from ${target.target} " +
+          s"with ${json.string.getBytes.length} bytes " +
+          s"in ${(endTs - startTs) / 1000.0}s"
+      )
       log.trace(s"retrieved cluster state: ${abbreviate(json.string)}")
 
       if (request.statusCode == 200) {

src/main/scala/uk/co/gresearch/spark/dgraph/connector/DgraphReader.scala

@@ -19,39 +19,49 @@ package uk.co.gresearch.spark.dgraph.connector
 import org.apache.spark.sql.{DataFrame, DataFrameReader}
 
 case class DgraphReader(reader: DataFrameReader) {
+
   /**
-   * Loads all triples of a Dgraph database into a DataFrame. Requires at least one target.
-   * Use triples(targets.head, targets.tail: _*) if you want to provide a Seq[String].
+   * Loads all triples of a Dgraph database into a DataFrame. Requires at least one target. Use triples(targets.head,
+   * targets.tail: _*) if you want to provide a Seq[String].
    *
-   * @param target  a target
-   * @param targets more targets
-   * @return triples DataFrame
+   * @param target
+   *   a target
+   * @param targets
+   *   more targets
+   * @return
+   *   triples DataFrame
    */
   def triples(target: String, targets: String*): DataFrame =
     reader
       .format(TriplesSource)
       .load(Seq(target) ++ targets: _*)
 
   /**
-   * Loads all edges of a Dgraph database into a DataFrame. Requires at least one target.
-   * Use edges(targets.head, targets.tail: _*) if want to provide a Seq[String].
+   * Loads all edges of a Dgraph database into a DataFrame. Requires at least one target. Use edges(targets.head,
+   * targets.tail: _*) if want to provide a Seq[String].
    *
-   * @param target  a target
-   * @param targets more targets
-   * @return edges DataFrame
+   * @param target
+   *   a target
+   * @param targets
+   *   more targets
+   * @return
+   *   edges DataFrame
    */
   def edges(target: String, targets: String*): DataFrame =
     reader
       .format(EdgesSource)
       .load(Seq(target) ++ targets: _*)
 
   /**
-   * Loads all nodes of a Dgraph database into a DataFrame. Requires at least one target.
-   * Use nodes(targets.head, targets.tail: _*) if you want to provide a Seq[String].
+   * Loads all nodes of a Dgraph database into a DataFrame. Requires at least one target. Use nodes(targets.head,
+   * targets.tail: _*) if you want to provide a Seq[String].
    *
-   * @param target  a target
-   * @param targets more targets
-   * @return nodes DataFrame
+   * @param target
+   *   a target
+   * @param targets
+   *   more targets
+   * @return
+   *   nodes DataFrame
    */
   def nodes(target: String, targets: String*): DataFrame =
     reader

src/main/scala/uk/co/gresearch/spark/dgraph/connector/Filter.scala

@@ -54,8 +54,10 @@ case class PredicateNameIs(name: String) extends PredicateNameIsIn(Set(name))
 
 // PredicateValueIsIn comes with two semantics: one that has to be intersected and one that cannot
 abstract class PredicateValueIsIn(val names: Set[String], val values: Set[Any]) extends Filter
-case class IntersectPredicateValueIsIn(override val names: Set[String], override val values: Set[Any]) extends PredicateValueIsIn(names, values)
-case class SinglePredicateValueIsIn(name: String, override val values: Set[Any]) extends PredicateValueIsIn(Set(name), values)
+case class IntersectPredicateValueIsIn(override val names: Set[String], override val values: Set[Any])
+    extends PredicateValueIsIn(names, values)
+case class SinglePredicateValueIsIn(name: String, override val values: Set[Any])
+    extends PredicateValueIsIn(Set(name), values)
 
 case class ObjectTypeIsIn(types: Set[String]) extends Filter
 case class ObjectValueIsIn(values: Set[Any]) extends Filter