Merge pull request #362 from marklogic/feature/marklogic-spark-api

Refactor: Extracted marklogic-spark-api

Author: rjrudin

build.gradle

@@ -1,8 +1,32 @@
+plugins {
+  id "java-library"
+  id "jacoco-report-aggregation"
+  id "org.sonarqube" version "5.1.0.4882"
+}
+
+// See https://docs.gradle.org/current/samples/sample_jvm_multi_project_with_code_coverage_standalone.html .
+// This isn't picking up jacoco coverage of all the subprojects. Will figure that out soon.
+dependencies {
+  jacocoAggregation project(':marklogic-langchain4j')
+  jacocoAggregation project(':marklogic-spark-api')
+  jacocoAggregation project(':marklogic-spark-langchain4j')
+  jacocoAggregation project(':marklogic-spark-connector')
+}
+
+sonar {
+  properties {
+    property "sonar.projectKey", "marklogic-spark"
+    property "sonar.host.url", "http://localhost:9000"
+  }
+}
+
 subprojects {
   apply plugin: "java-library"
+  apply plugin: "jacoco"
+  apply plugin: "org.sonarqube"
 
   group = "com.marklogic"
-  version '2.5-SNAPSHOT'
+  version "2.5-SNAPSHOT"
 
   java {
     sourceCompatibility = 11
@@ -19,9 +43,18 @@ subprojects {
 
   test {
     useJUnitPlatform()
+    finalizedBy jacocoTestReport
     testLogging {
       events 'started', 'passed', 'skipped', 'failed'
       exceptionFormat 'full'
     }
   }
+
+  // See https://docs.gradle.org/current/userguide/jacoco_plugin.html .
+  jacocoTestReport {
+    dependsOn test
+    reports {
+      xml.required = true
+    }
+  }
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.9-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.11-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

marklogic-spark-api/README.md

@@ -0,0 +1,2 @@
+This module defines the public API that marklogic-spark-langchain4j can depend on without depending on Spark or the
+connector itself.

marklogic-spark-api/build.gradle

@@ -0,0 +1,3 @@
+dependencies {
+  implementation "org.slf4j:jcl-over-slf4j:2.0.13"
+}

marklogic-spark-connector/src/main/java/com/marklogic/spark/ConnectorException.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/ConnectorException.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright © 2024 MarkLogic Corporation. All Rights Reserved.
+ */
+package com.marklogic.spark;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.stream.Stream;
+
+public abstract class Context implements Serializable {
+
+    protected final Map<String, String> properties;
+
+    protected Context(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    public final boolean hasOption(String... options) {
+        return Stream.of(options)
+            .anyMatch(option -> properties.get(option) != null && properties.get(option).trim().length() > 0);
+    }
+
+    public final String getStringOption(String option) {
+        return getStringOption(option, null);
+    }
+
+    public final String getStringOption(String option, String defaultValue) {
+        return hasOption(option) ? properties.get(option).trim() : defaultValue;
+    }
+
+    public final int getIntOption(String optionName, int defaultValue, int minimumValue) {
+        return (int) getNumericOption(optionName, defaultValue, minimumValue);
+    }
+
+    public final long getNumericOption(String optionName, long defaultValue, long minimumValue) {
+        try {
+            long value = this.getProperties().containsKey(optionName) ?
+                Long.parseLong(this.getProperties().get(optionName)) :
+                defaultValue;
+            if (value != defaultValue && value < minimumValue) {
+                throw new ConnectorException(String.format("The value of '%s' must be %d or greater.", getOptionNameForMessage(optionName), minimumValue));
+            }
+            return value;
+        } catch (NumberFormatException ex) {
+            throw new ConnectorException(String.format("The value of '%s' must be numeric.", getOptionNameForMessage(optionName)), ex);
+        }
+    }
+
+    public final boolean getBooleanOption(String option, boolean defaultValue) {
+        return hasOption(option) ? Boolean.parseBoolean(getStringOption(option)) : defaultValue;
+    }
+    
+    public final String getOptionNameForMessage(String option) {
+        return Util.getOptionNameForErrorMessage(option);
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+}

marklogic-spark-api/src/main/java/com/marklogic/spark/Context.java

@@ -3,7 +3,6 @@
  */
 package com.marklogic.spark;
 
-import com.marklogic.client.document.DocumentManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -73,21 +72,4 @@ static String getOptionNameForErrorMessage(String option) {
         String optionName = bundle.getString(option);
         return optionName != null && optionName.trim().length() > 0 ? optionName.trim() : option;
     }
-
-    static Set<DocumentManager.Metadata> getRequestedMetadata(ContextSupport context) {
-        Set<DocumentManager.Metadata> set = new HashSet<>();
-        if (context.hasOption(Options.READ_DOCUMENTS_CATEGORIES)) {
-            for (String category : context.getStringOption(Options.READ_DOCUMENTS_CATEGORIES).split(",")) {
-                if ("content".equalsIgnoreCase(category)) {
-                    continue;
-                }
-                if ("metadata".equalsIgnoreCase(category)) {
-                    set.add(DocumentManager.Metadata.ALL);
-                } else {
-                    set.add(DocumentManager.Metadata.valueOf(category.toUpperCase()));
-                }
-            }
-        }
-        return set;
-    }
 }

marklogic-spark-connector/src/main/java/com/marklogic/spark/Options.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/Options.java

@@ -3,8 +3,6 @@ plugins {
   id 'com.gradleup.shadow' version '8.3.3'
   id "com.marklogic.ml-gradle" version "5.0.0"
   id 'maven-publish'
-  id "jacoco"
-  id "org.sonarqube" version "5.1.0.4882"
 }
 
 configurations {
@@ -29,12 +27,14 @@ configurations.all {
 
 dependencies {
   // This is compileOnly as any environment this is used in will provide the Spark dependencies itself.
-  compileOnly ('org.apache.spark:spark-sql_2.12:' + sparkVersion) {
+  compileOnly('org.apache.spark:spark-sql_2.12:' + sparkVersion) {
     // Excluded from our ETL tool for size reasons, so excluded here as well to ensure we don't need it.
     exclude module: "rocksdbjni"
   }
 
-  shadowDependencies (project(":marklogic-langchain4j")) {
+  shadowDependencies project(":marklogic-spark-api")
+
+  shadowDependencies("com.marklogic:marklogic-client-api:7.0.0") {
     // The Java Client uses Jackson 2.15.2; Scala 3.4.x does not yet support that and will throw the following error:
     // Scala module 2.14.2 requires Jackson Databind version >= 2.14.0 and < 2.15.0 - Found jackson-databind version 2.15.2
     // So the 4 Jackson modules are excluded to allow for Spark's to be used.
@@ -43,7 +43,7 @@ dependencies {
   }
 
   // Required for converting JSON to XML. Using 2.15.2 to align with Spark 3.5.3.
-  shadowDependencies ("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:2.15.2") {
+  shadowDependencies("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:2.15.2") {
     // Not needed, as the modules in this group that this dependency depends on are all provided by Spark.
     exclude group: "com.fasterxml.jackson.core"
   }
@@ -52,26 +52,30 @@ dependencies {
   shadowDependencies 'com.squareup.okhttp3:okhttp:4.12.0'
 
   // Supports reading and writing RDF data.
-  shadowDependencies ("org.apache.jena:jena-arq:4.10.0") {
+  shadowDependencies("org.apache.jena:jena-arq:4.10.0") {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
   }
 
+  // Allows us to test the langchain4j-specific features without including them in the connector by default.
+  testImplementation project(":marklogic-langchain4j")
+  testImplementation project(":marklogic-spark-langchain4j")
+
   // Supports testing the embedder feature.
   testImplementation "dev.langchain4j:langchain4j-embeddings-all-minilm-l6-v2:0.35.0"
 
   // Needed for some XML operations that are far easier with JDOM2 than with DOM.
   shadowDependencies "org.jdom:jdom2:2.0.6.1"
 
-  testImplementation ('com.marklogic:ml-app-deployer:5.0.0') {
+  testImplementation('com.marklogic:ml-app-deployer:5.0.0') {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
 
     // Use the Java Client declared above.
     exclude module: "marklogic-client-api"
   }
 
-  testImplementation ('com.marklogic:marklogic-junit5:1.5.0') {
+  testImplementation('com.marklogic:marklogic-junit5:1.5.0') {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
 
@@ -85,26 +89,10 @@ dependencies {
 }
 
 test {
-  finalizedBy jacocoTestReport
   // Allows mlHost to override the value in gradle.properties, which the test plumbing will default to.
   environment "mlHost", mlHost
 }
 
-// See https://docs.gradle.org/current/userguide/jacoco_plugin.html .
-jacocoTestReport {
-  dependsOn test
-  reports {
-    xml.required = true
-  }
-}
-
-sonar {
-  properties {
-    property "sonar.projectKey", "marklogic-spark"
-    property "sonar.host.url", "http://localhost:9000"
-  }
-}
-
 task reloadTestData(type: com.marklogic.gradle.task.MarkLogicTask) {
   description = "Convenience task for clearing the test database and reloading the test data; only intended for a connector developer to use."
   doLast {

marklogic-spark-connector/src/main/java/com/marklogic/spark/Util.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/Util.java

@@ -5,19 +5,22 @@
 
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.DatabaseClientFactory;
+import com.marklogic.client.document.DocumentManager;
 import com.marklogic.client.extra.okhttpclient.OkHttpClientConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 
-public class ContextSupport implements Serializable {
+public class ContextSupport extends Context implements Serializable {
 
     protected static final Logger logger = LoggerFactory.getLogger(ContextSupport.class);
-    private final Map<String, String> properties;
     private final boolean configuratorWasAdded;
 
     // Java Client 6.5.0 has a bug in it (to be fixed in 6.5.1 or 6.6.0) where multiple threads that use a configurator
@@ -26,7 +29,7 @@ public class ContextSupport implements Serializable {
     private static final Object CLIENT_LOCK = new Object();
 
     public ContextSupport(Map<String, String> properties) {
-        this.properties = properties;
+        super(properties);
         this.configuratorWasAdded = addOkHttpConfiguratorIfNecessary();
     }
 
@@ -92,10 +95,6 @@ public final boolean optionExists(String option) {
         return value != null && value.trim().length() > 0;
     }
 
-    public final String getOptionNameForMessage(String option) {
-        return Util.getOptionNameForErrorMessage(option);
-    }
-
     private void parseConnectionString(String value, Map<String, String> connectionProps) {
         ConnectionString connectionString = new ConnectionString(value, getOptionNameForMessage("spark.marklogic.client.uri"));
         connectionProps.put(Options.CLIENT_USERNAME, connectionString.getUsername());
@@ -109,24 +108,6 @@ private void parseConnectionString(String value, Map<String, String> connectionP
         }
     }
 
-    public final int getIntOption(String optionName, int defaultValue, int minimumValue) {
-        return (int) getNumericOption(optionName, defaultValue, minimumValue);
-    }
-
-    public final long getNumericOption(String optionName, long defaultValue, long minimumValue) {
-        try {
-            long value = this.getProperties().containsKey(optionName) ?
-                Long.parseLong(this.getProperties().get(optionName)) :
-                defaultValue;
-            if (value != defaultValue && value < minimumValue) {
-                throw new ConnectorException(String.format("The value of '%s' must be %d or greater.", getOptionNameForMessage(optionName), minimumValue));
-            }
-            return value;
-        } catch (NumberFormatException ex) {
-            throw new ConnectorException(String.format("The value of '%s' must be numeric.", getOptionNameForMessage(optionName)), ex);
-        }
-    }
-
     /**
      * Only intended for "write" use cases; an error on "read" is always expected to be propagated to the user.
      *
@@ -141,30 +122,10 @@ public final boolean isDirectConnection() {
         return value != null && value.equalsIgnoreCase(DatabaseClient.ConnectionType.DIRECT.name());
     }
 
-    public final boolean hasOption(String... options) {
-        return Util.hasOption(this.properties, options);
-    }
-
-    public final String getStringOption(String option) {
-        return getStringOption(option, null);
-    }
-
-    public final String getStringOption(String option, String defaultValue) {
-        return hasOption(option) ? properties.get(option).trim() : defaultValue;
-    }
-
-    public final boolean getBooleanOption(String option, boolean defaultValue) {
-        return hasOption(option) ? Boolean.parseBoolean(getStringOption(option)) : defaultValue;
-    }
-
     public final boolean isStreamingFiles() {
         return "true".equalsIgnoreCase(getStringOption(Options.STREAM_FILES));
     }
 
-    public Map<String, String> getProperties() {
-        return properties;
-    }
-
     /**
      * @return true if a configurator was added
      */
@@ -197,4 +158,21 @@ private boolean addOkHttpConfiguratorIfNecessary() {
         }
         return false;
     }
+
+    public static Set<DocumentManager.Metadata> getRequestedMetadata(ContextSupport context) {
+        Set<DocumentManager.Metadata> set = new HashSet<>();
+        if (context.hasOption(Options.READ_DOCUMENTS_CATEGORIES)) {
+            for (String category : context.getStringOption(Options.READ_DOCUMENTS_CATEGORIES).split(",")) {
+                if ("content".equalsIgnoreCase(category)) {
+                    continue;
+                }
+                if ("metadata".equalsIgnoreCase(category)) {
+                    set.add(DocumentManager.Metadata.ALL);
+                } else {
+                    set.add(DocumentManager.Metadata.valueOf(category.toUpperCase()));
+                }
+            }
+        }
+        return set;
+    }
 }