Skip to content

Commit 831ecdc

Browse files
rjrudinrjrudin
committed
MLE-26460 Refactor: Cleaning up usage of QueryConfig
Made QueryConfig into a record so it's immutable, making it easier to understand. And passing it to the constructor of QueryBatcherImpl instead of having 6 separate args.
1 parent 408406c commit 831ecdc

File tree

4 files changed

+113
-107
lines changed

4 files changed

+113
-107
lines changed

‎marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/DataMovementManagerImpl.java‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,10 @@
11
/*
2-
* Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
2+
* Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
33
*/
44
package com.marklogic.client.datamovement.impl;
55

66
import com.marklogic.client.DatabaseClient;
77
import com.marklogic.client.DatabaseClientBuilder;
8-
import com.marklogic.client.DatabaseClientFactory;
98
import com.marklogic.client.datamovement.*;
109
import com.marklogic.client.impl.DatabaseClientImpl;
1110
import com.marklogic.client.io.marker.ContentHandle;
@@ -123,10 +122,8 @@ private QueryBatcher newQueryBatcherImpl(SearchQueryDefinition query) {
123122
QueryBatcherImpl queryBatcher = null;
124123
// preprocess the query if the effective version is at least 10.0-5
125124
if (Long.compareUnsigned(getServerVersion(), Long.parseUnsignedLong("10000500")) >= 0) {
126-
DataMovementServices.QueryConfig queryConfig = service.initConfig("POST", query);
127-
queryBatcher = new QueryBatcherImpl(query, this, queryConfig.forestConfig,
128-
queryConfig.serializedCtsQuery, queryConfig.filtered,
129-
queryConfig.maxDocToUriBatchRatio, queryConfig.defaultDocBatchSize, queryConfig.maxUriBatchSize);
125+
QueryConfig queryConfig = service.initConfig("POST", query);
126+
queryBatcher = new QueryBatcherImpl(query, this, queryConfig);
130127
} else {
131128
queryBatcher = new QueryBatcherImpl(query, this, getForestConfig());
132129
}

‎marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/DataMovementServices.java‎

Lines changed: 55 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
2+
* Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
33
*/
44
package com.marklogic.client.datamovement.impl;
55

@@ -23,7 +23,8 @@
2323
import java.util.List;
2424

2525
public class DataMovementServices {
26-
private static Logger logger = LoggerFactory.getLogger(DataMovementServices.class);
26+
27+
private static final Logger logger = LoggerFactory.getLogger(DataMovementServices.class);
2728

2829
private DatabaseClient client;
2930

@@ -36,59 +37,58 @@ public DataMovementServices setClient(DatabaseClient client) {
3637
return this;
3738
}
3839

39-
QueryConfig initConfig(String method, SearchQueryDefinition qdef) {
40-
logger.debug("initializing forest configuration with query");
41-
if (qdef == null) throw new IllegalArgumentException("null query definition");
42-
43-
JsonNode result = ((DatabaseClientImpl) this.client).getServices()
44-
.forestInfo(null, method, new RequestParameters(), qdef, new JacksonHandle())
45-
.get();
46-
// System.out.println(result.toPrettyString());
47-
48-
QueryConfig queryConfig = new QueryConfig();
49-
50-
try {
51-
ObjectMapper mapper = new ObjectMapper();
52-
JsonNode queryResult = result.get("query");
53-
if (queryResult != null && queryResult.isObject() && queryResult.has("ctsquery")) {
54-
queryConfig.serializedCtsQuery = mapper.writeValueAsString(queryResult);
55-
logger.debug("initialized query to: {}", queryConfig.serializedCtsQuery);
56-
}
57-
JsonNode filteredResult = result.get("filtered");
58-
if (filteredResult != null && filteredResult.isBoolean()) {
59-
queryConfig.filtered = filteredResult.asBoolean();
60-
logger.debug("initialized filtering to: {}", queryConfig.filtered.toString());
61-
}
62-
JsonNode maxDocToUriBatchRatio = result.get("maxDocToUriBatchRatio");
63-
if (maxDocToUriBatchRatio != null && maxDocToUriBatchRatio.isInt()) {
64-
queryConfig.maxDocToUriBatchRatio = maxDocToUriBatchRatio.asInt();
65-
logger.debug("initialized maxDocToUriBatchRatio to : {}", queryConfig.maxDocToUriBatchRatio);
66-
} else {
67-
queryConfig.maxDocToUriBatchRatio = -1;
68-
}
69-
JsonNode defaultDocBatchSize = result.get("defaultDocBatchSize");
70-
if (defaultDocBatchSize != null && defaultDocBatchSize.isInt()) {
71-
queryConfig.defaultDocBatchSize = defaultDocBatchSize.asInt();
72-
logger.debug("initialized defaultDocBatchSize to : {}", queryConfig.defaultDocBatchSize);
73-
} else {
74-
queryConfig.defaultDocBatchSize = -1;
75-
}
76-
JsonNode maxUriBatchSize = result.get("maxUriBatchSize");
77-
if (maxUriBatchSize != null && maxUriBatchSize.isInt()) {
78-
queryConfig.maxUriBatchSize = maxUriBatchSize.asInt();
79-
logger.debug("initialized maxUriBatchSize to : {}", queryConfig.maxUriBatchSize);
80-
} else {
81-
queryConfig.maxUriBatchSize = -1;
82-
}
83-
84-
} catch (JsonProcessingException e) {
85-
logger.error("failed to initialize query", e);
86-
}
87-
88-
queryConfig.forestConfig = makeForestConfig(result.has("forests") ? result.get("forests") : result);
89-
90-
return queryConfig;
91-
}
40+
QueryConfig initConfig(String method, SearchQueryDefinition qdef) {
41+
logger.debug("initializing forest configuration with query");
42+
if (qdef == null) throw new IllegalArgumentException("null query definition");
43+
44+
JsonNode result = ((DatabaseClientImpl) this.client).getServices()
45+
.forestInfo(null, method, new RequestParameters(), qdef, new JacksonHandle())
46+
.get();
47+
48+
JsonNode queryResult = result.get("query");
49+
50+
String serializedCtsQuery = null;
51+
if (queryResult != null && queryResult.isObject() && queryResult.has("ctsquery")) {
52+
try {
53+
serializedCtsQuery = new ObjectMapper().writeValueAsString(queryResult);
54+
logger.debug("initialized query to: {}", serializedCtsQuery);
55+
} catch (JsonProcessingException e) {
56+
logger.warn("Unable to serialize query result while initializing QueryBatcher; cause: {}", e.getMessage());
57+
}
58+
}
59+
60+
JsonNode filteredResult = result.get("filtered");
61+
Boolean filtered = null;
62+
if (filteredResult != null && filteredResult.isBoolean()) {
63+
filtered = filteredResult.asBoolean();
64+
logger.debug("initialized filtering to: {}", filtered);
65+
}
66+
67+
JsonNode maxDocToUriBatchRatioNode = result.get("maxDocToUriBatchRatio");
68+
int maxDocToUriBatchRatio = -1;
69+
if (maxDocToUriBatchRatioNode != null && maxDocToUriBatchRatioNode.isInt()) {
70+
maxDocToUriBatchRatio = maxDocToUriBatchRatioNode.asInt();
71+
logger.debug("initialized maxDocToUriBatchRatio to : {}", maxDocToUriBatchRatio);
72+
}
73+
74+
JsonNode defaultDocBatchSizeNode = result.get("defaultDocBatchSize");
75+
int defaultDocBatchSize = -1;
76+
if (defaultDocBatchSizeNode != null && defaultDocBatchSizeNode.isInt()) {
77+
defaultDocBatchSize = defaultDocBatchSizeNode.asInt();
78+
logger.debug("initialized defaultDocBatchSize to : {}", defaultDocBatchSize);
79+
}
80+
81+
JsonNode maxUriBatchSizeNode = result.get("maxUriBatchSize");
82+
int maxUriBatchSize = -1;
83+
if (maxUriBatchSizeNode != null && maxUriBatchSizeNode.isInt()) {
84+
maxUriBatchSize = maxUriBatchSizeNode.asInt();
85+
logger.debug("initialized maxUriBatchSize to : {}", maxUriBatchSize);
86+
}
87+
88+
ForestConfiguration forestConfig = makeForestConfig(result.has("forests") ? result.get("forests") : result);
89+
return new QueryConfig(serializedCtsQuery, forestConfig, filtered,
90+
maxDocToUriBatchRatio, defaultDocBatchSize, maxUriBatchSize);
91+
}
9292

9393
ForestConfigurationImpl readForestConfig() {
9494
logger.debug("initializing forest configuration");
@@ -183,12 +183,4 @@ private String generateJobId() {
183183
return UUID.randomUUID().toString();
184184
}
185185

186-
static class QueryConfig {
187-
String serializedCtsQuery;
188-
ForestConfiguration forestConfig;
189-
Boolean filtered;
190-
int maxDocToUriBatchRatio;
191-
int defaultDocBatchSize;
192-
int maxUriBatchSize;
193-
}
194186
}

‎marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/QueryBatcherImpl.java‎

Lines changed: 39 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
2+
* Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
33
*/
44
package com.marklogic.client.datamovement.impl;
55

@@ -42,16 +42,17 @@
4242
* startIterating, withForestConfig, and retry.
4343
*/
4444
public class QueryBatcherImpl extends BatcherImpl implements QueryBatcher {
45-
private static Logger logger = LoggerFactory.getLogger(QueryBatcherImpl.class);
45+
private static final Logger logger = LoggerFactory.getLogger(QueryBatcherImpl.class);
4646
private String queryMethod;
4747
private SearchQueryDefinition query;
48-
private SearchQueryDefinition originalQuery;
4948
private Boolean filtered;
5049
private Iterator<String> iterator;
5150
private boolean threadCountSet = false;
52-
private List<QueryBatchListener> urisReadyListeners = new ArrayList<>();
53-
private List<QueryFailureListener> failureListeners = new ArrayList<>();
54-
private List<QueryBatcherListener> jobCompletionListeners = new ArrayList<>();
51+
52+
private final List<QueryBatchListener> urisReadyListeners = new ArrayList<>();
53+
private final List<QueryFailureListener> failureListeners = new ArrayList<>();
54+
private final List<QueryBatcherListener> jobCompletionListeners = new ArrayList<>();
55+
5556
private QueryThreadPoolExecutor threadPool;
5657
private boolean consistentSnapshot = false;
5758
private final AtomicLong batchNumber = new AtomicLong(0);
@@ -61,51 +62,51 @@ public class QueryBatcherImpl extends BatcherImpl implements QueryBatcher {
6162
private Map<Forest,AtomicLong> forestResults = new HashMap<>();
6263
private Map<Forest,AtomicBoolean> forestIsDone = new HashMap<>();
6364
private Map<Forest, AtomicInteger> retryForestMap = new HashMap<>();
64-
private AtomicBoolean runJobCompletionListeners = new AtomicBoolean(false);
65+
66+
private final AtomicBoolean runJobCompletionListeners = new AtomicBoolean(false);
6567
private final Object lock = new Object();
6668
private final Map<Forest,List<QueryTask>> blackListedTasks = new HashMap<>();
69+
6770
private boolean isSingleThreaded = false;
71+
6872
private long maxUris = Long.MAX_VALUE;
6973
private long maxBatches = Long.MAX_VALUE;
7074
private int maxDocToUriBatchRatio;
7175
private int docToUriBatchRatio;
7276
private int defaultDocBatchSize;
7377
private int maxUriBatchSize;
7478

75-
QueryBatcherImpl(
76-
SearchQueryDefinition originalQuery, DataMovementManager moveMgr, ForestConfiguration forestConfig,
77-
String serializedCtsQuery, Boolean filtered, int maxDocToUriBatchRatio, int defaultDocBatchSize, int maxUriBatchSize
78-
) {
79-
this(moveMgr, forestConfig, maxDocToUriBatchRatio, defaultDocBatchSize, maxUriBatchSize);
80-
// TODO: skip conversion in DataMovementManagerImpl.newQueryBatcherImpl() unless canSerializeQueryAsJSON()
81-
if (serializedCtsQuery != null && serializedCtsQuery.length() > 0 &&
82-
originalQuery instanceof AbstractSearchQueryDefinition &&
83-
((AbstractSearchQueryDefinition) originalQuery).canSerializeQueryAsJSON()) {
84-
QueryManagerImpl queryMgr = (QueryManagerImpl) getPrimaryClient().newQueryManager();
85-
this.queryMethod = "POST";
86-
this.query = queryMgr.newRawCtsQueryDefinition(new StringHandle(serializedCtsQuery).withFormat(Format.JSON));
87-
this.originalQuery = originalQuery;
88-
if (filtered != null) {
89-
this.filtered = filtered;
90-
}
91-
} else {
92-
initQuery(originalQuery);
93-
}
94-
}
79+
QueryBatcherImpl(SearchQueryDefinition originalQuery, DataMovementManager moveMgr, QueryConfig queryConfig) {
80+
this(moveMgr, queryConfig);
81+
82+
final String serializedCtsQuery = queryConfig.serializedCtsQuery();
83+
if (serializedCtsQuery != null && !serializedCtsQuery.isEmpty() &&
84+
originalQuery instanceof AbstractSearchQueryDefinition &&
85+
((AbstractSearchQueryDefinition) originalQuery).canSerializeQueryAsJSON()) {
86+
QueryManagerImpl queryMgr = (QueryManagerImpl) getPrimaryClient().newQueryManager();
87+
this.queryMethod = "POST";
88+
this.query = queryMgr.newRawCtsQueryDefinition(new StringHandle(serializedCtsQuery).withFormat(Format.JSON));
89+
this.filtered = queryConfig.filtered();
90+
} else {
91+
initQuery(originalQuery);
92+
}
93+
}
94+
9595
public QueryBatcherImpl(SearchQueryDefinition query, DataMovementManager moveMgr, ForestConfiguration forestConfig) {
9696
this(moveMgr, forestConfig);
9797
initQuery(query);
9898
}
99+
99100
public QueryBatcherImpl(Iterator<String> iterator, DataMovementManager moveMgr, ForestConfiguration forestConfig) {
100101
this(moveMgr, forestConfig);
101102
this.iterator = iterator;
102103
}
103-
private QueryBatcherImpl(DataMovementManager moveMgr, ForestConfiguration forestConfig,
104-
int maxDocToUriBatchRatio, int defaultDocBatchSize, int maxUriBatchSize) {
105-
this(moveMgr, forestConfig);
106-
this.maxDocToUriBatchRatio = maxDocToUriBatchRatio;
107-
this.defaultDocBatchSize = defaultDocBatchSize;
108-
this.maxUriBatchSize = maxUriBatchSize;
104+
105+
private QueryBatcherImpl(DataMovementManager moveMgr, QueryConfig queryConfig) {
106+
this(moveMgr, queryConfig.forestConfig());
107+
this.maxDocToUriBatchRatio = queryConfig.maxDocToUriBatchRatio();
108+
this.defaultDocBatchSize = queryConfig.defaultDocBatchSize();
109+
this.maxUriBatchSize = queryConfig.maxUriBatchSize();
109110
withBatchSize(defaultDocBatchSize);
110111
}
111112
private QueryBatcherImpl(DataMovementManager moveMgr, ForestConfiguration forestConfig) {
@@ -187,7 +188,7 @@ public void retryWithFailureListeners(QueryEvent queryEvent) {
187188
}
188189

189190
private void retry(QueryEvent queryEvent, boolean callFailListeners) {
190-
if ( isStopped() == true ) {
191+
if ( isStopped()) {
191192
logger.warn("Job is now stopped, aborting the retry");
192193
return;
193194
}
@@ -449,7 +450,7 @@ public synchronized void start(JobTicket ticket) {
449450

450451
private synchronized void initialize() {
451452
Forest[] forests = getForestConfig().listForests();
452-
if ( threadCountSet == false ) {
453+
if ( !threadCountSet ) {
453454
if ( query != null ) {
454455
logger.warn("threadCount not set--defaulting to number of forests ({})", forests.length);
455456
withThreadCount(forests.length * docToUriBatchRatio);
@@ -529,7 +530,7 @@ public synchronized QueryBatcher withForestConfig(ForestConfiguration forestConf
529530
List<DatabaseClient> newClientList = clients(hostNames);
530531
clientList.set(newClientList);
531532
boolean started = (threadPool != null);
532-
if ( started == true && oldForests.size() > 0 ) calculateDeltas(oldForests, forests);
533+
if ( started && !oldForests.isEmpty() ) calculateDeltas(oldForests, forests);
533534
return this;
534535
}
535536

@@ -550,7 +551,7 @@ private synchronized void calculateDeltas(Set<Forest> oldForests, Forest[] fores
550551
// this forest is not black-listed
551552
blackListedForests.remove(forest);
552553
}
553-
if ( blackListedForests.size() > 0 ) {
554+
if ( !blackListedForests.isEmpty() ) {
554555
DataMovementManagerImpl moveMgrImpl = getMoveMgr();
555556
String primaryHost = moveMgrImpl.getPrimaryClient().getHost();
556557
if ( getHostNames(blackListedForests).contains(primaryHost) ) {
@@ -562,7 +563,7 @@ private synchronized void calculateDeltas(Set<Forest> oldForests, Forest[] fores
562563
}
563564

564565
private synchronized void cleanupExistingTasks(Set<Forest> addedForests, Set<Forest> restartedForests, Set<Forest> blackListedForests) {
565-
if ( blackListedForests.size() > 0 ) {
566+
if ( !blackListedForests.isEmpty() ) {
566567
logger.warn("removing jobs related to hosts [{}] from the queue", getHostNames(blackListedForests));
567568
// since some forests have been removed, let's remove from the queue any jobs that were targeting that forest
568569
List<Runnable> tasks = new ArrayList<>();

‎marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/QueryConfig.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
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.impl;
5+
6+
import com.marklogic.client.datamovement.ForestConfiguration;
7+
8+
record QueryConfig(
9+
String serializedCtsQuery,
10+
ForestConfiguration forestConfig,
11+
Boolean filtered,
12+
int maxDocToUriBatchRatio,
13+
int defaultDocBatchSize,
14+
int maxUriBatchSize
15+
) {
16+
}

0 commit comments

Comments
 (0)