MLE-27078 Incremental write now uses an unsignedLong

Not sure eval will be kept, but all 3 approaches are now properly using an unsignedLong.

Also fixed a performance issue in the view filter where "op.in" was being used instead of a "where" + documentQuery.

Author: rjrudin

Jenkinsfile

@@ -150,8 +150,8 @@ pipeline {
 				}
 			}
 			steps {
-				setupDockerMarkLogic("progressofficial/marklogic-db:latest")
-				
+				setupDockerMarkLogic("ml-docker-db-dev-tierpoint.bed-artifactory.bedford.progress.com/marklogic/marklogic-server-ubi:latest-12")
+
 				sh label: 'run marklogic-client-api tests', script: '''#!/bin/bash
 					export JAVA_HOME=$JAVA_HOME_DIR
 					export GRADLE_USER_HOME=$WORKSPACE/$GRADLE_DIR

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteEvalFilter.java

@@ -19,11 +19,17 @@
  */
 class IncrementalWriteEvalFilter extends IncrementalWriteFilter {
 
+	// The hash value is cast to a String based on this analysis from Copilot:
+	// "Since the field index is unsignedLong, cts.valueTuples returns an
+	// xs:unsignedLong value, which JavaScript represents as an IEEE 754
+	// double. Values above −1 (~9.2 quadrillion) would silently lose precision."
+	// It is then cast back to an unsignedLong when the value is retrieved
+	// from the JSON response object.
 	private static final String EVAL_SCRIPT = """
 		const tuples = cts.valueTuples([cts.uriReference(), cts.fieldReference(hashKeyName)], null, cts.documentQuery(uris));
 		const response = {};
 		for (var tuple of tuples) {
-		  response[tuple[0]] = tuple[1];
+		  response[tuple[0]] = String(tuple[1]);
 		}
 		response
 		""";
@@ -49,7 +55,7 @@ public DocumentWriteSet apply(DocumentWriteSetFilter.Context context) {
 
 			return filterDocuments(context, uri -> {
 				if (response.has(uri)) {
-					return response.get(uri).asText();
+					return Long.parseUnsignedLong(response.get(uri).asText());
 				}
 				return null;
 			});

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteFilter.java

@@ -227,7 +227,7 @@ public IncrementalWriteConfig getConfig() {
 		return config;
 	}
 
-	protected final DocumentWriteSet filterDocuments(Context context, Function<String, String> hashRetriever) {
+	protected final DocumentWriteSet filterDocuments(Context context, Function<String, Long> hashRetriever) {
 		final DocumentWriteSet newWriteSet = context.getDatabaseClient().newDocumentManager().newWriteSet();
 		final List<DocumentWriteOperation> skippedDocuments = new ArrayList<>();
 		final String timestamp = Instant.now().toString();
@@ -246,8 +246,8 @@ protected final DocumentWriteSet filterDocuments(Context context, Function<Strin
 				continue;
 			}
 
-			final String contentHash = computeHash(serializedContent);
-			final String existingHash = hashRetriever.apply(doc.getUri());
+			final long contentHash = computeHash(serializedContent);
+			final Long existingHash = hashRetriever.apply(doc.getUri());
 			if (logger.isTraceEnabled()) {
 				logger.trace("URI: {}, existing Hash: {}, new Hash: {}", doc.getUri(), existingHash, contentHash);
 			}
@@ -317,13 +317,12 @@ private boolean isPossiblyJsonContent(String content) {
 		return trimmed.startsWith("{") || trimmed.startsWith("[");
 	}
 
-	private String computeHash(String content) {
+	private long computeHash(String content) {
 		byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
-		long hash = hashFunction.hashBytes(bytes);
-		return Long.toHexString(hash);
+		return hashFunction.hashBytes(bytes);
 	}
 
-	protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation op, String hashKeyName, String hash,
+	protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation op, String hashKeyName, long hash,
 															  String timestampKeyName, String timestamp) {
 		DocumentMetadataHandle newMetadata = new DocumentMetadataHandle();
 		if (op.getMetadata() != null) {
@@ -335,7 +334,7 @@ protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation
 			newMetadata.getMetadataValues().putAll(originalMetadata.getMetadataValues());
 		}
 
-		newMetadata.getMetadataValues().put(hashKeyName, hash);
+		newMetadata.getMetadataValues().put(hashKeyName, Long.toUnsignedString(hash));
 		newMetadata.getMetadataValues().put(timestampKeyName, timestamp);
 
 		return new DocumentWriteOperationImpl(op.getUri(), newMetadata, op.getContent(), op.getTemporalDocumentURI());

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteOpticFilter.java

@@ -34,7 +34,7 @@ public DocumentWriteSet apply(Context context) {
 		RowTemplate rowTemplate = new RowTemplate(context.getDatabaseClient());
 
 		try {
-			Map<String, String> existingHashes = rowTemplate.query(op ->
+			Map<String, Long> existingHashes = rowTemplate.query(op ->
 					op.fromLexicons(Map.of(
 						"uri", op.cts.uriReference(),
 						"hash", op.cts.fieldReference(getConfig().getHashKeyName())
@@ -43,10 +43,10 @@ public DocumentWriteSet apply(Context context) {
 					),
 
 				rows -> {
-					Map<String, String> map = new HashMap<>();
+					Map<String, Long> map = new HashMap<>();
 					rows.forEach(row -> {
 						String uri = row.getString("uri");
-						String existingHash = row.getString("hash");
+						long existingHash = Long.parseUnsignedLong(row.getString("hash"));
 						map.put(uri, existingHash);
 					});
 					return map;

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteViewFilter.java

@@ -13,7 +13,10 @@
 
 /**
  * Uses an Optic query with fromView to get the existing hash values for a set of URIs from a TDE view.
- * This implementation requires a TDE template to be deployed that extracts the URI and hash metadata.
+ * This implementation requires a TDE template to be deployed that contains at minimum a "uri" column
+ * and a column matching the configured hash key name, plus any other columns desired.
+ * The query uses a {@code where} with a {@code cts.documentQuery} to filter rows by URI, which is
+ * significantly faster than filtering via {@code op.in}.
  *
  * @since 8.1.0
  */
@@ -33,20 +36,22 @@ public DocumentWriteSet apply(Context context) {
 		RowTemplate rowTemplate = new RowTemplate(context.getDatabaseClient());
 
 		try {
-			Map<String, String> existingHashes = rowTemplate.query(op ->
-					op.fromView(getConfig().getSchemaName(), getConfig().getViewName())
-						.where(op.in(op.col("uri"), op.xs.stringSeq(uris))),
-
+			Map<String, Long> existingHashes = rowTemplate.query(op ->
+					op.fromView(getConfig().getSchemaName(), getConfig().getViewName(), "")
+						.where(op.cts.documentQuery(op.xs.stringSeq(uris)))
+				,
 				rows -> {
-					Map<String, String> map = new HashMap<>();
+					Map<String, Long> map = new HashMap<>();
 					rows.forEach(row -> {
 						String uri = row.getString("uri");
-						String existingHash = row.getString("hash");
-						map.put(uri, existingHash);
+						String hashString = row.getString(getConfig().getHashKeyName());
+						if (hashString != null && !hashString.isEmpty()) {
+							long existingHash = Long.parseUnsignedLong(hashString);
+							map.put(uri, existingHash);
+						}
 					});
 					return map;
-				}
-			);
+				});
 
 			if (logger.isDebugEnabled()) {
 				logger.debug("Retrieved {} existing hashes for batch of size {}", existingHashes.size(), uris.length);

marklogic-client-api/src/test/java/com/marklogic/client/datamovement/filter/IncrementalWriteFilterTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
+ * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
  */
 package com.marklogic.client.datamovement.filter;
 
@@ -35,7 +35,7 @@ void addHashToMetadata() {
 		DocumentWriteOperation doc2 = new DocumentWriteOperationImpl("/2.xml", metadata, new StringHandle("<doc2/>"));
 
 		final String timestamp = Instant.now().toString();
-		doc2 = IncrementalWriteFilter.addHashToMetadata(doc2, "theField", "abc123", "theTimestamp", timestamp);
+		doc2 = IncrementalWriteFilter.addHashToMetadata(doc2, "theField", 12345, "theTimestamp", timestamp);
 
 		assertEquals(metadata, doc1.getMetadata(), "doc1 should still have the original metadata object");
 
@@ -46,7 +46,7 @@ void addHashToMetadata() {
 		assertEquals("value1", metadata2.getProperties().get("prop1"), "property should be preserved");
 
 		assertEquals("value1", metadata2.getMetadataValues().get("meta1"), "metadata value should be preserved");
-		assertEquals("abc123", metadata2.getMetadataValues().get("theField"), "hash field should be added");
+		assertEquals("12345", metadata2.getMetadataValues().get("theField"), "hash field should be added");
 		assertEquals(timestamp, metadata2.getMetadataValues().get("theTimestamp"), "timestamp should be added");
 	}
 }

marklogic-client-api/src/test/java/com/marklogic/client/datamovement/filter/IncrementalWriteTest.java

@@ -251,6 +251,50 @@ void fromView() {
 		verifyIncrementalWriteWorks();
 	}
 
+	@Test
+	void loadWithEvalFilterThenVerifyOpticFilterSkipsAll() {
+		filter = IncrementalWriteFilter.newBuilder()
+			.useEvalQuery(true)
+			.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
+			.build();
+
+		writeTenDocuments();
+		assertEquals(10, writtenCount.get());
+		assertEquals(0, skippedCount.get());
+
+		// Switch to the Optic fromLexicons filter.
+		writtenCount.set(0);
+		skippedCount.set(0);
+		filter = IncrementalWriteFilter.newBuilder()
+			.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
+			.build();
+
+		writeTenDocuments();
+		assertEquals(0, writtenCount.get(), "No documents should be written since the hashes stored by the eval " +
+			"filter should be recognized as unchanged by the Optic filter.");
+		assertEquals(10, skippedCount.get());
+	}
+
+	@Test
+	void loadWithOpticFilterThenVerifyEvalFilterSkipsAll() {
+		writeTenDocuments();
+		assertEquals(10, writtenCount.get());
+		assertEquals(0, skippedCount.get());
+
+		// Switch to the eval filter.
+		writtenCount.set(0);
+		skippedCount.set(0);
+		filter = IncrementalWriteFilter.newBuilder()
+			.useEvalQuery(true)
+			.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
+			.build();
+
+		writeTenDocuments();
+		assertEquals(0, writtenCount.get(), "No documents should be written since the hashes stored by the Optic " +
+			"filter should be recognized as unchanged by the eval filter.");
+		assertEquals(10, skippedCount.get());
+	}
+
 	@Test
 	void emptyValuesForFromView() {
 		filter = IncrementalWriteFilter.newBuilder()
@@ -332,8 +376,8 @@ private void verifyDocumentsHasHashInMetadataKey() {
 
 			String hash = metadata.getMetadataValues().get("incrementalWriteHash");
 			try {
-				// Can use Java's support for parsing unsigned longs in base 16 to verify the hash is valid.
-				Long.parseUnsignedLong(hash, 16);
+				// Verify the hash is a valid unsigned long in base 10 decimal.
+				Long.parseUnsignedLong(hash);
 			} catch (NumberFormatException e) {
 				fail("Document " + doc.getUri() + " has an invalid incrementalWriteHash value: " + hash);
 			}

test-app/src/main/ml-config/databases/content-database.json

@@ -224,15 +224,13 @@
 			"invalid-values": "reject"
 		},
 		{
-			"scalar-type": "string",
-			"collation": "http://marklogic.com/collation/",
+			"scalar-type": "unsignedLong",
 			"field-name": "incrementalWriteHash",
 			"range-value-positions": false,
 			"invalid-values": "reject"
 		},
 		{
-			"scalar-type": "string",
-			"collation": "http://marklogic.com/collation/",
+			"scalar-type": "unsignedLong",
 			"field-name": "myWriteHash",
 			"range-value-positions": false,
 			"invalid-values": "reject"

test-app/src/main/ml-schemas/tde/incrementalWriteHash.json

@@ -13,8 +13,8 @@
 						"val": "xdmp:node-uri(.)"
 					},
 					{
-						"name": "hash",
-						"scalarType": "string",
+						"name": "incrementalWriteHash",
+						"scalarType": "unsignedLong",
 						"val": "xdmp:node-metadata-value(., 'incrementalWriteHash')",
 						"nullable": true
 					}