Merge pull request #391 from marklogic/release/2.5.1

rjrudin · web-flow · commit 94993197e78a · 2025-01-07T13:35:27.000-05:00
Release/2.5.1
diff --git a/NOTICE.txt b/NOTICE.txt
@@ -14,7 +14,7 @@ jackson-dataformat-xml 2.15.2 (Apache-2.0)
 jdom2 2.0.6.1 (Apache-2.0)
 jena-arq 4.10.0 (Apache-2.0)
 langchain4j 0.35.0 (Apache-2.0)
-marklogic-client-api 7.0.0 (Apache-2.0)
+marklogic-client-api 7.1.0 (Apache-2.0)
 okhttp 4.12.0 (Apache-2.0)
 
 Common Licenses
@@ -23,7 +23,7 @@ Apache License 2.0 (Apache-2.0)
 
 Third-Party Components
 
-The following is a list of the third-party components used by the MarkLogic® Spark connector 2.5.0 (last updated December 17, 2024):
+The following is a list of the third-party components used by the MarkLogic® Spark connector 2.5.1 (last updated January 6, 2025):
 
 jackson-dataformat-xml 2.15.2 (Apache-2.0)
 https://repo1.maven.org/maven2/com/fasterxml/jackson/dataformat/jackson-dataformat-xml/
@@ -41,7 +41,7 @@ langchain4j 0.35.0 (Apache-2.0)
 https://repo1.maven.org/maven2/dev/langchain4j/langchain4j/
 For the full text of the Apache-2.0 license, see Apache License 2.0 (Apache-2.0)
 
-marklogic-client-api 7.0.0 (Apache-2.0)
+marklogic-client-api 7.1.0 (Apache-2.0)
 https://repo1.maven.org/maven2/com/marklogic/marklogic-client-api/
 For the full text of the Apache-2.0 license, see Apache License 2.0 (Apache-2.0)
 
diff --git a/build.gradle b/build.gradle
@@ -16,7 +16,7 @@ subprojects {
   apply plugin: "jacoco"
 
   group = "com.marklogic"
-  version "2.5.0"
+  version "2.5-SNAPSHOT"
 
   java {
     sourceCompatibility = 11
diff --git a/docs/getting-started/jupyter.md b/docs/getting-started/jupyter.md
@@ -32,15 +32,15 @@ connector and also to initialize Spark:
 
 ```
 import os
-os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars "/path/to/marklogic-spark-connector-2.5.0.jar" pyspark-shell'
+os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars "/path/to/marklogic-spark-connector-2.5.1.jar" pyspark-shell'
 
 from pyspark.sql import SparkSession
 spark = SparkSession.builder.master("local[*]").appName('My Notebook').getOrCreate()
 spark.sparkContext.setLogLevel("WARN")
 spark
 ```
 
-The path of `/path/to/marklogic-spark-connector-2.5.0.jar` should be changed to match the location of the connector 
+The path of `/path/to/marklogic-spark-connector-2.5.1.jar` should be changed to match the location of the connector 
 jar on your filesystem. You are free to customize the `spark` variable in any manner you see fit as well. 
 
 Now that you have an initialized Spark session, you can run any of the examples found in the 
diff --git a/docs/getting-started/pyspark.md b/docs/getting-started/pyspark.md
@@ -30,7 +30,7 @@ shell by pressing `ctrl-D`.
 
 Run PySpark from the directory that you downloaded the connector to per the [setup instructions](setup.md):
 
-    pyspark --jars marklogic-spark-connector-2.5.0.jar
+    pyspark --jars marklogic-spark-connector-2.5.1.jar
 
 The `--jars` command line option is PySpark's method for utilizing Spark connectors. Each Spark environment should have
 a similar mechanism for including third party connectors; please see the documentation for your particular Spark
diff --git a/docs/getting-started/setup.md b/docs/getting-started/setup.md
@@ -31,10 +31,10 @@ have an instance of MarkLogic running, you can skip step 4 below, but ensure tha
 extracted directory contains valid connection properties for your instance of MarkLogic.
 
 1. From [this repository's Releases page](https://github.com/marklogic/marklogic-spark-connector/releases), select 
-   the latest release and download the `marklogic-spark-getting-started-2.5.0.zip` file.
+   the latest release and download the `marklogic-spark-getting-started-2.5.1.zip` file.
 2. Extract the contents of the downloaded zip file. 
 3. Open a terminal window and go to the directory created by extracting the zip file; the directory should have a 
-   name of "marklogic-spark-getting-started-2.5.0".
+   name of "marklogic-spark-getting-started-2.5.1".
 4. Run `docker-compose up -d` to start an instance of MarkLogic
 5. Ensure that the `./gradlew` file is executable; depending on your operating system, you may need to run
    `chmod 755 gradlew` to make the file executable.
diff --git a/examples/entity-aggregation/build.gradle b/examples/entity-aggregation/build.gradle
@@ -8,7 +8,7 @@ repositories {
 
 dependencies {
   implementation 'org.apache.spark:spark-sql_2.12:3.5.3'
-  implementation "com.marklogic:marklogic-spark-connector:2.5.0"
+  implementation "com.marklogic:marklogic-spark-connector:2.5.1"
   implementation "org.postgresql:postgresql:42.7.4"
 }
 
diff --git a/examples/getting-started/marklogic-spark-getting-started.ipynb b/examples/getting-started/marklogic-spark-getting-started.ipynb
@@ -9,7 +9,7 @@
    "source": [
     "# Make the MarkLogic connector available to the underlying PySpark application.\n",
     "import os\n",
-    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars \"marklogic-spark-connector-2.5.0.jar\" pyspark-shell'\n",
+    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars \"marklogic-spark-connector-2.5.1.jar\" pyspark-shell'\n",
     "\n",
     "# Define the connection details for the getting-started example application.\n",
     "client_uri = \"spark-example-user:password@localhost:8003\"\n",
diff --git a/examples/java-dependency/build.gradle b/examples/java-dependency/build.gradle
@@ -8,7 +8,7 @@ repositories {
 
 dependencies {
   implementation 'org.apache.spark:spark-sql_2.12:3.5.3'
-  implementation 'com.marklogic:marklogic-spark-connector:2.5.0'
+  implementation 'com.marklogic:marklogic-spark-connector:2.5.1'
 }
 
 task runApp(type: JavaExec) {
diff --git a/marklogic-langchain4j/build.gradle b/marklogic-langchain4j/build.gradle
@@ -1,6 +1,6 @@
 dependencies {
   compileOnly "com.fasterxml.jackson.core:jackson-databind:2.17.2"
-  api "com.marklogic:marklogic-client-api:7.0.0"
+  api "com.marklogic:marklogic-client-api:7.1.0"
 
   // Supports splitting documents.
   api "dev.langchain4j:langchain4j:0.35.0"
diff --git a/marklogic-spark-api/src/main/java/com/marklogic/spark/Options.java b/marklogic-spark-api/src/main/java/com/marklogic/spark/Options.java
@@ -15,6 +15,12 @@ public abstract class Options {
     public static final String CLIENT_URI = "spark.marklogic.client.uri";
     public static final String CLIENT_USERNAME = "spark.marklogic.client.username";
 
+    /**
+     * Alias for "spark.marklogic.client.uri", which will be deprecated soon in favor of this better name.
+     * @since 2.5.1
+     */
+    public static final String CLIENT_CONNECTION_STRING = "spark.marklogic.client.connectionString";
+
     public static final String READ_INVOKE = "spark.marklogic.read.invoke";
     public static final String READ_JAVASCRIPT = "spark.marklogic.read.javascript";
     public static final String READ_JAVASCRIPT_FILE = "spark.marklogic.read.javascriptFile";
diff --git a/marklogic-spark-connector/build.gradle b/marklogic-spark-connector/build.gradle
@@ -25,7 +25,7 @@ dependencies {
 
   shadowDependencies project(":marklogic-spark-api")
 
-  shadowDependencies("com.marklogic:marklogic-client-api:7.0.0") {
+  shadowDependencies("com.marklogic:marklogic-client-api:7.1.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.
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/ConnectionString.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/ConnectionString.java
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/ContextSupport.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/ContextSupport.java
@@ -77,8 +77,8 @@ protected final Map<String, String> buildConnectionProperties() {
         connectionProps.put("spark.marklogic.client.authType", "digest");
         connectionProps.put("spark.marklogic.client.connectionType", "gateway");
         connectionProps.putAll(getProperties());
-        if (optionExists(Options.CLIENT_URI)) {
-            parseConnectionString(getProperties().get(Options.CLIENT_URI), connectionProps);
+        if (connectionProps.containsKey(Options.CLIENT_URI)) {
+            connectionProps.put(Options.CLIENT_CONNECTION_STRING, connectionProps.get(Options.CLIENT_URI));
         }
         if ("true".equalsIgnoreCase(getProperties().get(Options.CLIENT_SSL_ENABLED))) {
             connectionProps.put("spark.marklogic.client.sslProtocol", "default");
@@ -91,19 +91,6 @@ public final boolean optionExists(String option) {
         return value != null && value.trim().length() > 0;
     }
 
-    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());
-        connectionProps.put(Options.CLIENT_PASSWORD, connectionString.getPassword());
-        connectionProps.put(Options.CLIENT_HOST, connectionString.getHost());
-        connectionProps.put(Options.CLIENT_PORT, connectionString.getPort() + "");
-
-        String db = connectionString.getDatabase();
-        if (db != null && db.trim().length() > 0) {
-            connectionProps.put(Options.CLIENT_DATABASE, db);
-        }
-    }
-
     /**
      * Only intended for "write" use cases; an error on "read" is always expected to be propagated to the user.
      *
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/DefaultSource.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/DefaultSource.java
@@ -9,6 +9,7 @@
 import com.marklogic.spark.reader.document.DocumentRowSchema;
 import com.marklogic.spark.reader.document.DocumentTable;
 import com.marklogic.spark.reader.file.TripleRowSchema;
+import com.marklogic.spark.reader.optic.OpticReadContext;
 import com.marklogic.spark.reader.optic.SchemaInferrer;
 import com.marklogic.spark.writer.WriteContext;
 import org.apache.spark.sql.SparkSession;
@@ -86,7 +87,14 @@ public Table getTable(StructType schema, Transform[] partitioning, Map<String, S
             ReadProgressLogger.initialize(readProgressInterval, "Triples read: {}");
             return new DocumentTable(TripleRowSchema.SCHEMA);
         } else if (properties.get(Options.READ_OPTIC_QUERY) != null) {
-            ReadProgressLogger.initialize(readProgressInterval, "Rows read: {}");
+            final long batchSize = tempContext.getNumericOption(Options.READ_BATCH_SIZE, OpticReadContext.DEFAULT_BATCH_SIZE, 0);
+            if (readProgressInterval > 0 && batchSize > readProgressInterval) {
+                Util.MAIN_LOGGER.info("Batch size is greater than interval for reading progress, so will use batch size of {} to log progress instead.",
+                    batchSize);
+                ReadProgressLogger.initialize(batchSize, "Rows read: {}");
+            } else {
+                ReadProgressLogger.initialize(readProgressInterval, "Rows read: {}");
+            }
             return new MarkLogicTable(schema, properties);
         } else if (Util.isReadWithCustomCodeOperation(properties)) {
             ReadProgressLogger.initialize(readProgressInterval, "Items read: {}");
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/customcode/CustomCodeScan.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/customcode/CustomCodeScan.java
@@ -4,6 +4,8 @@
 package com.marklogic.spark.reader.customcode;
 
 import com.marklogic.client.DatabaseClient;
+import com.marklogic.client.eval.EvalResultIterator;
+import com.marklogic.client.eval.ServerEvaluationCall;
 import com.marklogic.spark.ConnectorException;
 import com.marklogic.spark.Options;
 import org.apache.spark.sql.connector.read.Batch;
@@ -26,14 +28,13 @@ public CustomCodeScan(CustomCodeContext customCodeContext) {
 
         if (this.customCodeContext.hasPartitionCode()) {
             DatabaseClient client = this.customCodeContext.connectToMarkLogic();
-            try {
-                this.customCodeContext
-                    .buildCall(client, new CustomCodeContext.CallOptions(
-                        Options.READ_PARTITIONS_INVOKE, Options.READ_PARTITIONS_JAVASCRIPT, Options.READ_PARTITIONS_XQUERY,
-                        Options.READ_PARTITIONS_JAVASCRIPT_FILE, Options.READ_PARTITIONS_XQUERY_FILE
-                    ))
-                    .eval()
-                    .forEach(result -> this.partitions.add(result.getString()));
+            ServerEvaluationCall call = this.customCodeContext
+                .buildCall(client, new CustomCodeContext.CallOptions(
+                    Options.READ_PARTITIONS_INVOKE, Options.READ_PARTITIONS_JAVASCRIPT, Options.READ_PARTITIONS_XQUERY,
+                    Options.READ_PARTITIONS_JAVASCRIPT_FILE, Options.READ_PARTITIONS_XQUERY_FILE
+                ));
+            try (EvalResultIterator iter = call.eval()) {
+                iter.forEach(result -> this.partitions.add(result.getString()));
             } catch (Exception ex) {
                 throw new ConnectorException(String.format("Unable to retrieve partitions; cause: %s", ex.getMessage()), ex);
             } finally {
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/optic/OpticReadContext.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/optic/OpticReadContext.java
@@ -39,7 +39,7 @@ public class OpticReadContext extends ContextSupport {
     // choose to perform multiple operations on the dataset, each of which may benefit from a fairly different batch
     // size. 100k has been chosen as the default batch size to strike a reasonable balance for operations that do need
     // to collect all the rows, such as writing the dataset to another data source.
-    private static final long DEFAULT_BATCH_SIZE = 100000;
+    public static final long DEFAULT_BATCH_SIZE = 100000;
 
     private PlanAnalysis planAnalysis;
     private StructType schema;
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/customcode/CustomCodeWriter.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/customcode/CustomCodeWriter.java
@@ -38,12 +38,18 @@ class CustomCodeWriter implements DataWriter<InternalRow> {
     private final String externalVariableDelimiter;
     private ObjectMapper objectMapper;
 
+    // Only used for logging.
+    private final int partitionId;
+    private final long taskId;
+
     // Updated after each call to MarkLogic.
     private int successItemCount;
     private int failedItemCount;
 
-    CustomCodeWriter(CustomCodeContext customCodeContext) {
+    CustomCodeWriter(CustomCodeContext customCodeContext, int partitionId, long taskId) {
         this.customCodeContext = customCodeContext;
+        this.partitionId = partitionId;
+        this.taskId = taskId;
         this.databaseClient = customCodeContext.connectToMarkLogic();
         this.jsonRowSerializer = new JsonRowSerializer(customCodeContext.getSchema(), customCodeContext.getProperties());
 
@@ -103,9 +109,6 @@ private void flush() {
             return;
         }
 
-        if (logger.isTraceEnabled()) {
-            logger.trace("Calling custom code in MarkLogic");
-        }
         final int itemCount = currentBatch.size();
         ServerEvaluationCall call = customCodeContext.buildCall(
             this.databaseClient,
@@ -148,6 +151,10 @@ private ArrayNode makeArrayFromCurrentBatch() {
     }
 
     private void executeCall(ServerEvaluationCall call, int itemCount) {
+        if (Util.MAIN_LOGGER.isDebugEnabled()) {
+            // Helps ensure that the proper number of partitions and tasks/threads are being used by Spark.
+            Util.MAIN_LOGGER.debug("Processing items, count: {}; partition: {}; task: {}", itemCount, partitionId, taskId);
+        }
         try {
             call.evalAs(String.class);
             this.successItemCount += itemCount;
diff --git a/marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/customcode/CustomCodeWriterFactory.java b/marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/customcode/CustomCodeWriterFactory.java
@@ -19,11 +19,11 @@ public CustomCodeWriterFactory(CustomCodeContext customCodeContext) {
 
     @Override
     public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
-        return new CustomCodeWriter(customCodeContext);
+        return new CustomCodeWriter(customCodeContext, partitionId, taskId);
     }
 
     @Override
     public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long epochId) {
-        return new CustomCodeWriter(customCodeContext);
+        return new CustomCodeWriter(customCodeContext, partitionId, taskId);
     }
 }
diff --git a/marklogic-spark-connector/src/test/java/com/marklogic/spark/BuildConnectionPropertiesTest.java b/marklogic-spark-connector/src/test/java/com/marklogic/spark/BuildConnectionPropertiesTest.java
@@ -38,10 +38,9 @@ void clientUriWithDatabase() {
             "database in the connection string, then the database value specified separately should still be used. " +
             "This avoids a potential issue where the user is not aware that the connection string accepts a database " +
             "and thus specifies it via the database option.");
-        assertEquals("user", connectionProps.get(Options.CLIENT_USERNAME));
-        assertEquals("password", connectionProps.get(Options.CLIENT_PASSWORD));
-        assertEquals("host", connectionProps.get(Options.CLIENT_HOST));
-        assertEquals("8016", connectionProps.get(Options.CLIENT_PORT));
+        assertEquals("user:password@host:8016", connectionProps.get(Options.CLIENT_CONNECTION_STRING),
+            "The new connection string option in 2.5.1 is intended to be used going forward as the better name. " +
+                "For now, the client.uri option is simply aliased to the new option.");
     }
 
     @Test
diff --git a/marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/optic/ReadRowsTest.java b/marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/optic/ReadRowsTest.java
diff --git a/marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/optic/ReadWithClientUriTest.java b/marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/optic/ReadWithClientUriTest.java
diff --git a/marklogic-spark-connector/src/test/java/com/marklogic/spark/writer/customcode/ProcessWithCustomCodeTest.java b/marklogic-spark-connector/src/test/java/com/marklogic/spark/writer/customcode/ProcessWithCustomCodeTest.java

Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,7 @@ repositories {`
`8`	`8`
`9`	`9`	`dependencies {`
`10`	`10`	`implementation 'org.apache.spark:spark-sql_2.12:3.5.3'`
`11`		`- implementation "com.marklogic:marklogic-spark-connector:2.5.0"`
	`11`	`+ implementation "com.marklogic:marklogic-spark-connector:2.5.1"`
`12`	`12`	`implementation "org.postgresql:postgresql:42.7.4"`
`13`	`13`	`}`
`14`	`14`