Skip to content

Commit 7aaf7e1

Browse files
XAVIER VANDEMEULEBROUCKEakonczak
XAVIER VANDEMEULEBROUCKE
authored and
akonczak
committed
DATAES-158 - DATAES-159
1 parent 455df72 commit 7aaf7e1

File tree

3 files changed

+111
-34
lines changed

3 files changed

+111
-34
lines changed

‎src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchOperations.java

Lines changed: 12 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,16 +15,16 @@
1515
*/
1616
package org.springframework.data.elasticsearch.core;
1717

18-
import java.util.LinkedList;
19-
import java.util.List;
20-
import java.util.Map;
21-
import java.util.Set;
22-
2318
import org.elasticsearch.action.update.UpdateResponse;
2419
import org.springframework.data.domain.Page;
2520
import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;
2621
import org.springframework.data.elasticsearch.core.query.*;
2722

23+
import java.util.LinkedList;
24+
import java.util.List;
25+
import java.util.Map;
26+
import java.util.Set;
27+
2828
/**
2929
* ElasticsearchOperations
3030
*
@@ -323,6 +323,13 @@ public interface ElasticsearchOperations {
323323
*/
324324
void bulkIndex(List<IndexQuery> queries);
325325

326+
/**
327+
* Bulk update all objects. Will do update
328+
*
329+
* @param queries
330+
*/
331+
void bulkUpdate(List<UpdateQuery> queries);
332+
326333
/**
327334
* Delete the one object with provided id
328335
*

‎src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplate.java

Lines changed: 55 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -15,21 +15,6 @@
1515
*/
1616
package org.springframework.data.elasticsearch.core;
1717

18-
import static org.apache.commons.collections.CollectionUtils.isNotEmpty;
19-
import static org.apache.commons.lang.StringUtils.*;
20-
import static org.elasticsearch.action.search.SearchType.*;
21-
import static org.elasticsearch.client.Requests.*;
22-
import static org.elasticsearch.cluster.metadata.AliasAction.Type.*;
23-
import static org.elasticsearch.common.collect.Sets.*;
24-
import static org.elasticsearch.index.VersionType.*;
25-
import static org.springframework.data.elasticsearch.core.MappingBuilder.*;
26-
27-
import java.io.BufferedReader;
28-
import java.io.IOException;
29-
import java.io.InputStreamReader;
30-
import java.lang.reflect.Method;
31-
import java.util.*;
32-
3318
import org.apache.commons.collections.CollectionUtils;
3419
import org.elasticsearch.action.ListenableActionFuture;
3520
import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
@@ -95,6 +80,23 @@
9580
import org.springframework.data.mapping.PersistentProperty;
9681
import org.springframework.util.Assert;
9782

83+
import java.io.BufferedReader;
84+
import java.io.IOException;
85+
import java.io.InputStreamReader;
86+
import java.lang.reflect.Method;
87+
import java.util.*;
88+
89+
import static org.apache.commons.collections.CollectionUtils.isNotEmpty;
90+
import static org.apache.commons.lang.StringUtils.isBlank;
91+
import static org.apache.commons.lang.StringUtils.isNotBlank;
92+
import static org.elasticsearch.action.search.SearchType.SCAN;
93+
import static org.elasticsearch.client.Requests.indicesExistsRequest;
94+
import static org.elasticsearch.client.Requests.refreshRequest;
95+
import static org.elasticsearch.cluster.metadata.AliasAction.Type.ADD;
96+
import static org.elasticsearch.common.collect.Sets.newHashSet;
97+
import static org.elasticsearch.index.VersionType.EXTERNAL;
98+
import static org.springframework.data.elasticsearch.core.MappingBuilder.buildMapping;
99+
98100
/**
99101
* ElasticsearchTemplate
100102
*
@@ -447,23 +449,35 @@ public String index(IndexQuery query) {
447449

448450
@Override
449451
public UpdateResponse update(UpdateQuery query) {
452+
return this.prepareUpdate(query).execute().actionGet();
453+
}
454+
455+
private UpdateRequestBuilder prepareUpdate(UpdateQuery query) {
450456
String indexName = isNotBlank(query.getIndexName()) ? query.getIndexName() : getPersistentEntityFor(query.getClazz()).getIndexName();
451457
String type = isNotBlank(query.getType()) ? query.getType() : getPersistentEntityFor(query.getClazz()).getIndexType();
452458
Assert.notNull(indexName, "No index defined for Query");
453459
Assert.notNull(type, "No type define for Query");
454460
Assert.notNull(query.getId(), "No Id define for Query");
455461
Assert.notNull(query.getUpdateRequest(), "No IndexRequest define for Query");
456462
UpdateRequestBuilder updateRequestBuilder = client.prepareUpdate(indexName, type, query.getId());
457-
if (query.DoUpsert()) {
458-
updateRequestBuilder.setDocAsUpsert(true)
459-
.setDoc(query.getUpdateRequest().doc())
463+
464+
if(query.getUpdateRequest().script() == null) {
465+
// doc
466+
if (query.DoUpsert()) {
467+
updateRequestBuilder.setDocAsUpsert(true)
468+
.setDoc(query.getUpdateRequest().doc());
469+
} else {
470+
updateRequestBuilder.setDoc(query.getUpdateRequest().doc());
471+
}
472+
} else {
473+
// or script
474+
updateRequestBuilder
460475
.setScript(query.getUpdateRequest().script(), query.getUpdateRequest().scriptType())
461476
.setScriptParams(query.getUpdateRequest().scriptParams())
462477
.setScriptLang(query.getUpdateRequest().scriptLang());
463-
} else {
464-
updateRequestBuilder.setDoc(query.getUpdateRequest().doc());
465478
}
466-
return updateRequestBuilder.execute().actionGet();
479+
480+
return updateRequestBuilder;
467481
}
468482

469483
@Override
@@ -486,6 +500,26 @@ public void bulkIndex(List<IndexQuery> queries) {
486500
}
487501
}
488502

503+
@Override
504+
public void bulkUpdate(List<UpdateQuery> queries) {
505+
BulkRequestBuilder bulkRequest = client.prepareBulk();
506+
for (UpdateQuery query : queries) {
507+
bulkRequest.add(prepareUpdate(query));
508+
}
509+
BulkResponse bulkResponse = bulkRequest.execute().actionGet();
510+
if (bulkResponse.hasFailures()) {
511+
Map<String, String> failedDocuments = new HashMap<String, String>();
512+
for (BulkItemResponse item : bulkResponse.getItems()) {
513+
if (item.isFailed())
514+
failedDocuments.put(item.getId(), item.getFailureMessage());
515+
}
516+
throw new ElasticsearchException(
517+
"Bulk indexing has failures. Use ElasticsearchException.getFailedDocuments() for detailed messages ["
518+
+ failedDocuments + "]", failedDocuments
519+
);
520+
}
521+
}
522+
489523
@Override
490524
public <T> boolean indexExists(Class<T> clazz) {
491525
return indexExists(getPersistentEntityFor(clazz).getIndexName());

‎src/test/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplateTests.java

Lines changed: 44 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,6 @@
1515
*/
1616
package org.springframework.data.elasticsearch.core;
1717

18-
import static org.apache.commons.lang.RandomStringUtils.*;
19-
import static org.elasticsearch.index.query.FilterBuilders.*;
20-
import static org.elasticsearch.index.query.QueryBuilders.*;
21-
import static org.hamcrest.Matchers.*;
22-
import static org.junit.Assert.*;
23-
24-
import java.util.*;
25-
2618
import org.apache.commons.lang.StringUtils;
2719
import org.elasticsearch.action.get.MultiGetItemResponse;
2820
import org.elasticsearch.action.get.MultiGetResponse;
@@ -53,6 +45,15 @@
5345
import org.springframework.test.context.ContextConfiguration;
5446
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
5547

48+
import java.util.*;
49+
50+
import static org.apache.commons.lang.RandomStringUtils.randomNumeric;
51+
import static org.elasticsearch.index.query.FilterBuilders.boolFilter;
52+
import static org.elasticsearch.index.query.FilterBuilders.termFilter;
53+
import static org.elasticsearch.index.query.QueryBuilders.*;
54+
import static org.hamcrest.Matchers.*;
55+
import static org.junit.Assert.*;
56+
5657
/**
5758
* @author Rizwan Idrees
5859
* @author Mohsin Husen
@@ -254,6 +255,41 @@ public void shouldDoBulkIndex() {
254255
assertThat(sampleEntities.getTotalElements(), is(equalTo(2L)));
255256
}
256257

258+
259+
@Test
260+
public void shouldDoBulkUpdate() {
261+
//given
262+
String documentId = randomNumeric(5);
263+
String messageBeforeUpdate = "some test message";
264+
String messageAfterUpdate = "test message";
265+
266+
SampleEntity sampleEntity = new SampleEntityBuilder(documentId)
267+
.message(messageBeforeUpdate)
268+
.version(System.currentTimeMillis()).build();
269+
270+
IndexQuery indexQuery = getIndexQuery(sampleEntity);
271+
272+
elasticsearchTemplate.index(indexQuery);
273+
elasticsearchTemplate.refresh(SampleEntity.class, true);
274+
275+
IndexRequest indexRequest = new IndexRequest();
276+
indexRequest.source("message", messageAfterUpdate);
277+
UpdateQuery updateQuery = new UpdateQueryBuilder().withId(documentId)
278+
.withClass(SampleEntity.class).withIndexRequest(indexRequest).build();
279+
280+
281+
List<UpdateQuery> queries = new ArrayList<UpdateQuery>();
282+
queries.add(updateQuery);
283+
284+
// when
285+
elasticsearchTemplate.bulkUpdate(queries);
286+
//then
287+
GetQuery getQuery = new GetQuery();
288+
getQuery.setId(documentId);
289+
SampleEntity indexedEntity = elasticsearchTemplate.queryForObject(getQuery, SampleEntity.class);
290+
assertThat(indexedEntity.getMessage(), is(messageAfterUpdate));
291+
}
292+
257293
@Test
258294
public void shouldDeleteDocumentForGivenId() {
259295
// given

0 commit comments

Comments
 (0)