MLE-26918 Added fromView support for incremental write

The plan is to dump eval support and just offer fromLexicons and fromView, but need to test out fromView a bit first.

Author: rjrudin

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

@@ -53,6 +53,8 @@ public static class Builder {
 		private String[] jsonExclusions;
 		private String[] xmlExclusions;
 		private Map<String, String> xmlNamespaces;
+		private String schemaName;
+		private String viewName;
 
 		/**
 		 * @param keyName the name of the MarkLogic metadata key that will hold the hash value; defaults to "incrementalWriteHash".
@@ -128,9 +130,36 @@ public Builder xmlNamespaces(Map<String, String> namespaces) {
 			return this;
 		}
 
+		/**
+		 * Configures the filter to use a TDE view for retrieving hash values instead of field range indexes.
+		 * This approach requires a TDE template to be deployed that extracts the URI and hash metadata.
+		 *
+		 * @param schemaName the schema name of the TDE view
+		 * @param viewName   the view name of the TDE view
+		 * @return this builder
+		 */
+		public Builder fromView(String schemaName, String viewName) {
+			boolean schemaEmpty = schemaName == null || schemaName.trim().isEmpty();
+			boolean viewEmpty = viewName == null || viewName.trim().isEmpty();
+
+			if (schemaEmpty && !viewEmpty) {
+				throw new IllegalArgumentException("Schema name cannot be null or empty when view name is provided");
+			}
+			if (!schemaEmpty && viewEmpty) {
+				throw new IllegalArgumentException("View name cannot be null or empty when schema name is provided");
+			}
+
+			this.schemaName = schemaName;
+			this.viewName = viewName;
+			return this;
+		}
+
 		public IncrementalWriteFilter build() {
 			validateJsonExclusions();
 			validateXmlExclusions();
+			if (schemaName != null && viewName != null) {
+				return new IncrementalWriteViewFilter(hashKeyName, timestampKeyName, canonicalizeJson, skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces, schemaName, viewName);
+			}
 			if (useEvalQuery) {
 				return new IncrementalWriteEvalFilter(hashKeyName, timestampKeyName, canonicalizeJson, skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces);
 			}

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
+ */
+package com.marklogic.client.datamovement.filter;
+
+import com.marklogic.client.FailedRequestException;
+import com.marklogic.client.document.DocumentWriteOperation;
+import com.marklogic.client.document.DocumentWriteSet;
+import com.marklogic.client.row.RowTemplate;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+/**
+ * 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.
+ *
+ * @since 8.1.0
+ */
+class IncrementalWriteViewFilter extends IncrementalWriteFilter {
+
+	private final String schemaName;
+	private final String viewName;
+
+	IncrementalWriteViewFilter(String hashKeyName, String timestampKeyName, boolean canonicalizeJson,
+							   Consumer<DocumentWriteOperation[]> skippedDocumentsConsumer, String[] jsonExclusions,
+							   String[] xmlExclusions, Map<String, String> xmlNamespaces,
+							   String schemaName, String viewName) {
+		super(hashKeyName, timestampKeyName, canonicalizeJson, skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces);
+		this.schemaName = schemaName;
+		this.viewName = viewName;
+	}
+
+	@Override
+	public DocumentWriteSet apply(Context context) {
+		final String[] uris = context.getDocumentWriteSet().stream()
+			.filter(op -> DocumentWriteOperation.OperationType.DOCUMENT_WRITE.equals(op.getOperationType()))
+			.map(DocumentWriteOperation::getUri)
+			.toArray(String[]::new);
+
+		RowTemplate rowTemplate = new RowTemplate(context.getDatabaseClient());
+
+		try {
+			Map<String, String> existingHashes = rowTemplate.query(op ->
+					op.fromView(schemaName, viewName)
+						.where(op.in(op.col("uri"), op.xs.stringSeq(uris))),
+
+				rows -> {
+					Map<String, String> map = new HashMap<>();
+					rows.forEach(row -> {
+						String uri = row.getString("uri");
+						String existingHash = row.getString("hash");
+						map.put(uri, existingHash);
+					});
+					return map;
+				}
+			);
+
+			return filterDocuments(context, uri -> existingHashes.get(uri));
+		} catch (FailedRequestException e) {
+			String message = "Unable to query for existing incremental write hashes from view " + schemaName + "." + viewName + "; cause: " + e.getMessage();
+			throw new FailedRequestException(message, e.getFailedRequest());
+		}
+	}
+}

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

@@ -228,6 +228,57 @@ void binaryDocument() {
 			"expected. Exclusions cannot be specified for them.");
 	}
 
+	@Test
+	void fromView() {
+		filter = IncrementalWriteFilter.newBuilder()
+			.fromView("javaClient", "incrementalWriteHash")
+			.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
+			.build();
+
+		verifyIncrementalWriteWorks();
+	}
+
+	@Test
+	void emptyValuesForFromView() {
+		filter = IncrementalWriteFilter.newBuilder()
+			// Empty/null values are ignored, as long as both schema/view are empty/null. This makes life a little
+			// easier for a connector in that the connector does not need to check for empty/null values.
+			.fromView("", null)
+			.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
+			.build();
+
+		verifyIncrementalWriteWorks();
+	}
+
+	@Test
+	void invalidSchemaArg() {
+		IncrementalWriteFilter.Builder builder = IncrementalWriteFilter.newBuilder();
+		IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, () -> builder.fromView(null, "theView"));
+		assertEquals("Schema name cannot be null or empty when view name is provided", ex.getMessage());
+	}
+
+	@Test
+	void invalidViewArg() {
+		IncrementalWriteFilter.Builder builder = IncrementalWriteFilter.newBuilder();
+		IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, () -> builder.fromView("javaClient", null));
+		assertEquals("View name cannot be null or empty when schema name is provided", ex.getMessage());
+	}
+
+	@Test
+	void invalidView() {
+		filter = IncrementalWriteFilter.newBuilder()
+			.fromView("javaClient", "this-view-doesnt-exist")
+			.build();
+
+		writeTenDocuments();
+
+		assertNotNull(batchFailure.get());
+		String message = batchFailure.get().getMessage();
+		assertTrue(message.contains("Unable to query for existing incremental write hashes") && message.contains("SQL-TABLENOTFOUND"),
+			"When the user tries to use the incremental write feature with an invalid view, " +
+				"we should fail with a helpful error message. Actual message: " + message);
+	}
+
 	private void verifyIncrementalWriteWorks() {
 		writeTenDocuments();
 		verifyDocumentsHasHashInMetadataKey();

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

@@ -0,0 +1,25 @@
+{
+	"template": {
+		"description": "For incremental write that uses op.fromView instead of op.fromLexicons",
+		"context": "/doc",
+		"rows": [
+			{
+				"schemaName": "javaClient",
+				"viewName": "incrementalWriteHash",
+				"columns": [
+					{
+						"name": "uri",
+						"scalarType": "string",
+						"val": "xdmp:node-uri(.)"
+					},
+					{
+						"name": "hash",
+						"scalarType": "string",
+						"val": "xdmp:node-metadata-value(., 'incrementalWriteHash')",
+						"nullable": true
+					}
+				]
+			}
+		]
+	}
+}