Skip to content

Commit 6b8b217

Browse files
rjrudinrjrudin
committed
MLE-32904 Incremental write now supportings matching on a field
Supports a use case where the document to be matched against has had its URI modified since it was ingested.
1 parent 1d19510 commit 6b8b217

9 files changed

Lines changed: 312 additions & 12 deletions

File tree

‎marklogic-client-api/build.gradle‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
*/
44

55
plugins {
6+
id 'net.saliman.properties' version '1.6.0'
67
id 'maven-publish'
78
}
89

@@ -186,6 +187,19 @@ publishing {
186187
}
187188
}
188189
repositories {
190+
// Supports publishing to a GH packages repository for GH Action workflows that are not able to access our
191+
// internal repository.
192+
if (project.hasProperty("ghToken") && project.hasProperty("ghPackagesUrl")) {
193+
maven {
194+
name = "GitHubPackages"
195+
url = uri(ghPackagesUrl)
196+
credentials {
197+
username = ghActor
198+
password = ghToken
199+
}
200+
}
201+
}
202+
189203
maven {
190204
if (project.hasProperty("mavenUser")) {
191205
credentials {

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

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
public class IncrementalWriteConfig {
1818

1919
private final String hashKeyName;
20+
private final String sourceUriKeyName;
2021
private final String timestampKeyName;
2122
private final boolean canonicalizeJson;
2223
private final Consumer<DocumentWriteOperation[]> skippedDocumentsConsumer;
@@ -26,11 +27,24 @@ public class IncrementalWriteConfig {
2627
private final String schemaName;
2728
private final String viewName;
2829

29-
public IncrementalWriteConfig(String hashKeyName, String timestampKeyName, boolean canonicalizeJson,
30+
// This was mistakenly exposed as a public constructor but there's no need for clients to access it, while clients
31+
// should be able to access this class and its getters.
32+
@Deprecated(since = "8.3.0", forRemoval = true)
33+
public IncrementalWriteConfig(String hashKeyName, String timestampKeyName,
34+
boolean canonicalizeJson,
3035
Consumer<DocumentWriteOperation[]> skippedDocumentsConsumer,
3136
String[] jsonExclusions, String[] xmlExclusions, Map<String, String> xmlNamespaces,
3237
String schemaName, String viewName) {
38+
this(hashKeyName, null, timestampKeyName, canonicalizeJson, skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces, schemaName, viewName);
39+
}
40+
41+
IncrementalWriteConfig(String hashKeyName, String sourceUriKeyName, String timestampKeyName,
42+
boolean canonicalizeJson,
43+
Consumer<DocumentWriteOperation[]> skippedDocumentsConsumer,
44+
String[] jsonExclusions, String[] xmlExclusions, Map<String, String> xmlNamespaces,
45+
String schemaName, String viewName) {
3346
this.hashKeyName = hashKeyName;
47+
this.sourceUriKeyName = sourceUriKeyName;
3448
this.timestampKeyName = timestampKeyName;
3549
this.canonicalizeJson = canonicalizeJson;
3650
this.skippedDocumentsConsumer = skippedDocumentsConsumer;
@@ -45,6 +59,13 @@ public String getHashKeyName() {
4559
return hashKeyName;
4660
}
4761

62+
/**
63+
* @since 8.3.0
64+
*/
65+
public String getSourceUriKeyName() {
66+
return sourceUriKeyName;
67+
}
68+
4869
public String getTimestampKeyName() {
4970
return timestampKeyName;
5071
}

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

Lines changed: 52 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ public static Builder newBuilder() {
4646
public static class Builder {
4747

4848
private String hashKeyName = "incrementalWriteHash";
49+
private String sourceUriKeyName;
4950
private String timestampKeyName;
5051
private boolean canonicalizeJson = true;
5152
private Consumer<DocumentWriteOperation[]> skippedDocumentsConsumer;
@@ -75,6 +76,26 @@ public Builder timestampKeyName(String keyName) {
7576
return this;
7677
}
7778

79+
/**
80+
* Configures this filter to match existing documents by a stable "source URI" metadata
81+
* value instead of the document's actual/current URI. Useful when a document may be
82+
* relocated to a different URI after this filter writes it, since matching by URI alone
83+
* would otherwise stop working. If set, the filter will expect a field range index to exist on the
84+
* specified metadata key. The field range index will be utilized to find matching documents instead of a
85+
* document query on URIs.
86+
*
87+
* @param keyName the name of the MarkLogic metadata key that will hold each document's
88+
* source URI; defaults to null, which means matching is based on the
89+
* document's actual URI, as it has always worked.
90+
* @since 8.3.0
91+
*/
92+
public Builder sourceUriKeyName(String keyName) {
93+
if (keyName != null && !keyName.trim().isEmpty()) {
94+
this.sourceUriKeyName = keyName;
95+
}
96+
return this;
97+
}
98+
7899
/**
79100
* @param canonicalizeJson whether to canonicalize JSON content before hashing; defaults to true.
80101
* Delegates to https://github.com/erdtman/java-json-canonicalization for canonicalization.
@@ -146,8 +167,9 @@ public Builder fromView(String schemaName, String viewName) {
146167
public IncrementalWriteFilter build() {
147168
validateJsonExclusions();
148169
validateXmlExclusions();
149-
IncrementalWriteConfig config = new IncrementalWriteConfig(hashKeyName, timestampKeyName, canonicalizeJson,
150-
skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces, schemaName, viewName);
170+
IncrementalWriteConfig config = new IncrementalWriteConfig(hashKeyName, sourceUriKeyName, timestampKeyName,
171+
canonicalizeJson, skippedDocumentsConsumer, jsonExclusions, xmlExclusions, xmlNamespaces,
172+
schemaName, viewName);
151173

152174
if (schemaName != null && viewName != null) {
153175
return new IncrementalWriteFromViewFilter(config);
@@ -240,14 +262,16 @@ protected final DocumentWriteSet filterDocuments(Context context, Function<Strin
240262

241263
if (existingHash != null) {
242264
if (!existingHash.equals(contentHash)) {
243-
newWriteSet.add(addHashToMetadata(doc, config.getHashKeyName(), contentHash, config.getTimestampKeyName(), timestamp));
265+
newWriteSet.add(addHashToMetadata(doc, config.getHashKeyName(), config.getSourceUriKeyName(),
266+
contentHash, config.getTimestampKeyName(), timestamp));
244267
} else if (config.getSkippedDocumentsConsumer() != null) {
245268
skippedDocuments.add(doc);
246269
} else {
247270
// No consumer, so skip the document silently.
248271
}
249272
} else {
250-
newWriteSet.add(addHashToMetadata(doc, config.getHashKeyName(), contentHash, config.getTimestampKeyName(), timestamp));
273+
newWriteSet.add(addHashToMetadata(doc, config.getHashKeyName(), config.getSourceUriKeyName(),
274+
contentHash, config.getTimestampKeyName(), timestamp));
251275
}
252276
}
253277

@@ -310,6 +334,24 @@ private long computeHash(String content) {
310334

311335
protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation op, String hashKeyName, long hash,
312336
String timestampKeyName, String timestamp) {
337+
return addHashToMetadata(op, hashKeyName, null, hash, timestampKeyName, timestamp);
338+
}
339+
340+
/**
341+
* @param op the write operation to add hash (and optionally source-URI) metadata to.
342+
* @param hashKeyName the metadata key that will hold the content hash.
343+
* @param sourceUriKeyName the metadata key that will hold the document's source URI, or null if this
344+
* feature is not in use. If the operation's existing metadata already has a
345+
* value for this key, it is preserved rather than overwritten with
346+
* {@code op.getUri()}.
347+
* @param hash the newly computed content hash.
348+
* @param timestampKeyName the metadata key that will hold the write timestamp, or null if timestamps
349+
* are not in use.
350+
* @param timestamp the timestamp to store, if {@code timestampKeyName} is not null.
351+
*/
352+
protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation op, String hashKeyName,
353+
String sourceUriKeyName, long hash,
354+
String timestampKeyName, String timestamp) {
313355
DocumentMetadataHandle newMetadata = new DocumentMetadataHandle();
314356
if (op.getMetadata() != null) {
315357
DocumentMetadataHandle originalMetadata = (DocumentMetadataHandle) op.getMetadata();
@@ -324,6 +366,12 @@ protected static DocumentWriteOperation addHashToMetadata(DocumentWriteOperation
324366
if (timestampKeyName != null && !timestampKeyName.trim().isEmpty()) {
325367
newMetadata.getMetadataValues().put(timestampKeyName, timestamp);
326368
}
369+
if (sourceUriKeyName != null && !sourceUriKeyName.trim().isEmpty()
370+
&& !newMetadata.getMetadataValues().containsKey(sourceUriKeyName)) {
371+
// Only stamp this automatically if the caller hasn't already supplied an explicit value -
372+
// e.g. via metadata set directly on the incoming write operation.
373+
newMetadata.getMetadataValues().put(sourceUriKeyName, op.getUri());
374+
}
327375

328376
return new DocumentWriteOperationImpl(op.getUri(), newMetadata, op.getContent(), op.getTemporalDocumentURI());
329377
}

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,10 @@
1313

1414
/**
1515
* Uses an Optic fromLexicons query that depends on a field range index to retrieve URIs and
16-
* hash values.
16+
* hash values. When {@code sourceUriKeyName} is configured (see
17+
* {@link IncrementalWriteFilter.Builder#sourceUriKeyName(String)}), the "URI" side of the
18+
* lookup is instead backed by a field over that metadata key, since the matched document may
19+
* have since been relocated to a different physical URI.
1720
*
1821
* @since 8.1.0
1922
*/
@@ -26,14 +29,19 @@ class IncrementalWriteFromLexiconsFilter extends IncrementalWriteFilter {
2629
@Override
2730
public DocumentWriteSet apply(Context context) {
2831
final String[] uris = getUrisInBatch(context.getDocumentWriteSet());
32+
final String sourceUriKeyName = getConfig().getSourceUriKeyName();
2933

3034
try {
3135
Map<String, Long> existingHashes = new RowTemplate(context.getDatabaseClient()).query(op ->
3236
op.fromLexicons(Map.of(
33-
"uri", op.cts.uriReference(),
37+
"uri", sourceUriKeyName != null
38+
? op.cts.fieldReference(sourceUriKeyName)
39+
: op.cts.uriReference(),
3440
"hash", op.cts.fieldReference(getConfig().getHashKeyName())
3541
)).where(
36-
op.cts.documentQuery(op.xs.stringSeq(uris))
42+
sourceUriKeyName != null
43+
? op.cts.fieldRangeQuery(op.xs.string(sourceUriKeyName), op.xs.string("="), op.xs.stringSeq(uris))
44+
: op.cts.documentQuery(op.xs.stringSeq(uris))
3745
),
3846

3947
rows -> {

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

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import com.marklogic.client.FailedRequestException;
77
import com.marklogic.client.document.DocumentWriteSet;
8+
import com.marklogic.client.expression.PlanBuilder;
89
import com.marklogic.client.row.RowTemplate;
910

1011
import java.util.HashMap;
@@ -28,10 +29,16 @@ public DocumentWriteSet apply(Context context) {
2829
final String[] uris = getUrisInBatch(context.getDocumentWriteSet());
2930

3031
try {
31-
Map<String, Long> existingHashes = new RowTemplate(context.getDatabaseClient()).query(op ->
32-
op.fromView(getConfig().getSchemaName(), getConfig().getViewName(), "")
33-
.where(op.cts.documentQuery(op.xs.stringSeq(uris)))
34-
,
32+
Map<String, Long> existingHashes = new RowTemplate(context.getDatabaseClient()).query(op -> {
33+
PlanBuilder.ModifyPlan plan = op.fromView(getConfig().getSchemaName(), getConfig().getViewName(), "");
34+
final String sourceUriName = getConfig().getSourceUriKeyName();
35+
if (sourceUriName != null && !sourceUriName.trim().isEmpty()) {
36+
plan = plan.where(op.cts.fieldRangeQuery(op.xs.string(sourceUriName), op.xs.string("="), op.xs.stringSeq(uris)));
37+
} else {
38+
plan = plan.where(op.cts.documentQuery(op.xs.stringSeq(uris)));
39+
}
40+
return plan;
41+
},
3542
rows -> {
3643
Map<String, Long> map = new HashMap<>();
3744
rows.forEach(row -> {

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

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,4 +49,38 @@ void addHashToMetadata() {
4949
assertEquals("12345", metadata2.getMetadataValues().get("theField"), "hash field should be added");
5050
assertEquals(timestamp, metadata2.getMetadataValues().get("theTimestamp"), "timestamp should be added");
5151
}
52+
53+
@Test
54+
void addHashToMetadataStampsSourceUriWhenConfigured() {
55+
DocumentWriteOperation doc = new DocumentWriteOperationImpl("/source/1.json", null, new StringHandle("{}"));
56+
57+
doc = IncrementalWriteFilter.addHashToMetadata(doc, "theField", "sourceUri", 12345, null, null);
58+
59+
DocumentMetadataHandle metadata = (DocumentMetadataHandle) doc.getMetadata();
60+
assertEquals("/source/1.json", metadata.getMetadataValues().get("sourceUri"),
61+
"the document's own URI should be stamped as the source URI when none was already set");
62+
}
63+
64+
@Test
65+
void addHashToMetadataPreservesExplicitSourceUri() {
66+
DocumentMetadataHandle metadata = new DocumentMetadataHandle()
67+
.withMetadataValue("sourceUri", "/original/source.json");
68+
DocumentWriteOperation doc = new DocumentWriteOperationImpl("/relocated/1.json", metadata, new StringHandle("{}"));
69+
70+
doc = IncrementalWriteFilter.addHashToMetadata(doc, "theField", "sourceUri", 12345, null, null);
71+
72+
DocumentMetadataHandle newMetadata = (DocumentMetadataHandle) doc.getMetadata();
73+
assertEquals("/original/source.json", newMetadata.getMetadataValues().get("sourceUri"),
74+
"an explicit source URI already present in metadata should not be overwritten");
75+
}
76+
77+
@Test
78+
void addHashToMetadataWithoutSourceUriKeyNameDoesNotStampAnything() {
79+
DocumentWriteOperation doc = new DocumentWriteOperationImpl("/1.json", null, new StringHandle("{}"));
80+
81+
doc = IncrementalWriteFilter.addHashToMetadata(doc, "theField", null, 12345, null, null);
82+
83+
DocumentMetadataHandle metadata = (DocumentMetadataHandle) doc.getMetadata();
84+
assertEquals(1, metadata.getMetadataValues().size(), "only the hash field should be present");
85+
}
5286
}

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

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/*
2+
* Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
3+
*/
4+
package com.marklogic.client.datamovement.filter;
5+
6+
import com.marklogic.client.document.DocumentWriteOperation;
7+
import com.marklogic.client.impl.DocumentWriteOperationImpl;
8+
import com.marklogic.client.io.DocumentMetadataHandle;
9+
import com.marklogic.client.io.StringHandle;
10+
import com.marklogic.client.test.Common;
11+
import org.junit.jupiter.api.BeforeEach;
12+
import org.junit.jupiter.api.Test;
13+
14+
import java.util.ArrayList;
15+
import java.util.List;
16+
17+
import static org.junit.jupiter.api.Assertions.assertEquals;
18+
import static org.junit.jupiter.api.Assertions.assertNotNull;
19+
import static org.junit.jupiter.api.Assertions.assertNull;
20+
import static org.junit.jupiter.api.Assertions.assertTrue;
21+
22+
class IncrementalWriteSourceUriKeyNameTest extends AbstractIncrementalWriteTest {
23+
24+
private static final String SOURCE_URI_KEY_NAME = "incrementalWriteSourceUri";
25+
private static final String SOURCE_URI = "/incremental/test/source-uri-doc.xml";
26+
private static final String RELOCATED_URI = "/incremental/test/relocated-doc.xml";
27+
28+
@Override
29+
@BeforeEach
30+
void setup() {
31+
super.setup();
32+
filter = IncrementalWriteFilter.newBuilder()
33+
.sourceUriKeyName(SOURCE_URI_KEY_NAME)
34+
.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
35+
.build();
36+
}
37+
38+
@Test
39+
void stampsSourceUriMetadataAutomatically() {
40+
writeDoc(SOURCE_URI, "<doc>original content</doc>");
41+
42+
DocumentMetadataHandle metadata = Common.client.newDocumentManager().readMetadata(SOURCE_URI,
43+
new DocumentMetadataHandle());
44+
assertEquals(SOURCE_URI, metadata.getMetadataValues().get(SOURCE_URI_KEY_NAME),
45+
"The filter should automatically stamp the configured metadata key with the document's own URI.");
46+
}
47+
48+
@Test
49+
void matchesAfterExternalRelocation() {
50+
// Initial write via the filter.
51+
writeDoc(SOURCE_URI, "<doc>original content</doc>");
52+
assertEquals(1, writtenCount.get());
53+
assertEquals(0, skippedCount.get());
54+
55+
relocateDocument();
56+
57+
// Same source URI, unchanged content - should be recognized as unchanged even though the document
58+
// now physically lives at a different URI, because the lookup matches on the source-URI field value.
59+
writeDoc(SOURCE_URI, "<doc>original content</doc>");
60+
assertEquals(1, writtenCount.get(), "No new write should have occurred since the document was skipped.");
61+
assertEquals(1, skippedCount.get());
62+
63+
assertNull(Common.client.newDocumentManager().exists(SOURCE_URI),
64+
"Nothing should have been written back to the source URI since the write was skipped.");
65+
assertNotNull(Common.client.newDocumentManager().exists(RELOCATED_URI),
66+
"The relocated document should still be the only copy of this content.");
67+
}
68+
69+
@Test
70+
void changedContentIsWrittenAfterExternalRelocation() {
71+
writeDoc(SOURCE_URI, "<doc>original content</doc>");
72+
relocateDocument();
73+
74+
// Content has changed, so the write should go through - to the source URI, not the relocated one.
75+
writeDoc(SOURCE_URI, "<doc>modified content</doc>");
76+
assertEquals(2, writtenCount.get());
77+
assertEquals(0, skippedCount.get());
78+
79+
String newContent = Common.client.newTextDocumentManager().read(SOURCE_URI, new StringHandle()).get();
80+
assertNotNull(newContent);
81+
assertTrue(newContent.contains("<doc>modified content</doc>"));
82+
83+
DocumentMetadataHandle metadata = Common.client.newDocumentManager().readMetadata(SOURCE_URI,
84+
new DocumentMetadataHandle());
85+
assertEquals(SOURCE_URI, metadata.getMetadataValues().get(SOURCE_URI_KEY_NAME),
86+
"The rewritten document should have the source-URI metadata value stamped again.");
87+
}
88+
89+
private void writeDoc(String uri, String content) {
90+
List<DocumentWriteOperation> ops = new ArrayList<>();
91+
ops.add(new DocumentWriteOperationImpl(uri, METADATA, new StringHandle(content)));
92+
writeDocs(ops);
93+
}
94+
95+
/**
96+
* Simulates a process outside of this filter relocating the document to a new URI while preserving its
97+
* metadata (in particular, the source-URI metadata value), then removing the original.
98+
*/
99+
private void relocateDocument() {
100+
DocumentMetadataHandle metadata = Common.client.newDocumentManager().readMetadata(SOURCE_URI,
101+
new DocumentMetadataHandle());
102+
String content = Common.client.newTextDocumentManager().read(SOURCE_URI, new StringHandle()).get();
103+
104+
Common.client.newTextDocumentManager().write(RELOCATED_URI, metadata, new StringHandle(content));
105+
Common.client.newDocumentManager().delete(SOURCE_URI);
106+
}
107+
}

0 commit comments

Comments
 (0)