@Test public void nestedWithExtraQuery() throws Exception { AbstractClient client = ElasticTestUtil.getDevClient(); Elasticsearch elasticsearch = new Elasticsearch(); ESRequestMapper mapper = new ESRequestMapper(client, elasticsearch.getRequestMapping()); SyncData data = SyncDataTestUtil.update("role", "role").setId(1234L); data.setRepo("nested3").setEntity("nested3").addField("title", "b").addField("user_id", 2L); data.extraQuery("user", "user").filter("_id", -1L).select("username").addField("username"); data.esScriptUpdate(Filter.fieldId("roles.id")).mergeToNestedById("roles", "title", "user_id", "username"); Object builder = mapper.map(data); assertEquals("", "{\n" + " \"size\" : 1000,\n" + " \"query\" : {\n" + " \"bool\" : {\n" + " \"filter\" : [\n" + " {\n" + " \"nested\" : {\n" + " \"query\" : {\n" + " \"bool\" : {\n" + " \"filter\" : [\n" + " {\n" + " \"term\" : {\n" + " \"roles.id\" : {\n" + " \"value\" : 1234,\n" + " \"boost\" : 1.0\n" + " }\n" + " }\n" + " }\n" + " ],\n" + " \"disable_coord\" : false,\n" + " \"adjust_pure_negative\" : true,\n" + " \"boost\" : 1.0\n" + " }\n" + " },\n" + " \"path\" : \"roles\",\n" + " \"ignore_unmapped\" : false,\n" + " \"score_mode\" : \"avg\",\n" + " \"boost\" : 1.0\n" + " }\n" + " }\n" + " ],\n" + " \"disable_coord\" : false,\n" + " \"adjust_pure_negative\" : true,\n" + " \"boost\" : 1.0\n" + " }\n" + " }\n" + "}", ((AbstractBulkByScrollRequestBuilder)builder).source().toString()); assertEquals("", "update-by-query [nested3] updated with Script{type=inline, lang='painless', idOrCode='def target = ctx._source.roles.find(e -> e.id.equals(params.id));if (target != null) { target.user_id = params.user_id;target.title = params.title;target.username = params.username;}', options={}, params={id=1234, title=b, user_id=2, username=null}}", ((UpdateByQueryRequestBuilder) builder).request().toString()); }
@ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override public Object map(SyncData data) { esQueryMapper.parseExtraQueryContext(data.getExtraQueryContext()); StandardEvaluationContext context = data.getContext(); String index = eval(indexExpr, context); String type = eval(typeExpr, context); String id = data.getId() == null ? null : data.getId().toString(); switch (data.getType()) { case WRITE: if (esRequestMapping.getNoUseIdForIndex()) { return client.prepareIndex(index, type).setSource(requestBodyMapper.map(data)); } return client.prepareIndex(index, type, id).setSource(requestBodyMapper.map(data)); case DELETE: logger.info("Deleting doc from Elasticsearch, may affect performance"); if (id != null) { return client.prepareDelete(index, type, id); } logger.warn("Deleting doc by query, may affect performance"); return DeleteByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)); case UPDATE: if (id != null) { if (needScript(data)) { HashMap<String, Object> map = requestBodyMapper.map(data); UpdateRequestBuilder builder = client.prepareUpdate(index, type, id) .setScript(getScript(data, map)) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); if (esRequestMapping.isUpsert()) { builder.setUpsert(getUpsert(data)).setScriptedUpsert(true); } return builder; } else { return client.prepareUpdate(index, type, id).setDoc(requestBodyMapper.map(data)) .setDocAsUpsert(esRequestMapping.isUpsert()) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); } } else { logger.warn("Updating doc by query, may affect performance"); return UpdateByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)) .script(getScript(data, data.getFields())); } default: throw new IllegalArgumentException("Unsupported row event type: " + data); } }
ESRequestMapper implements Mapper<SyncData, Object> { @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override public Object map(SyncData data) { esQueryMapper.parseExtraQueryContext(data.getExtraQueryContext()); StandardEvaluationContext context = data.getContext(); String index = eval(indexExpr, context); String type = eval(typeExpr, context); String id = data.getId() == null ? null : data.getId().toString(); switch (data.getType()) { case WRITE: if (esRequestMapping.getNoUseIdForIndex()) { return client.prepareIndex(index, type).setSource(requestBodyMapper.map(data)); } return client.prepareIndex(index, type, id).setSource(requestBodyMapper.map(data)); case DELETE: logger.info("Deleting doc from Elasticsearch, may affect performance"); if (id != null) { return client.prepareDelete(index, type, id); } logger.warn("Deleting doc by query, may affect performance"); return DeleteByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)); case UPDATE: if (id != null) { if (needScript(data)) { HashMap<String, Object> map = requestBodyMapper.map(data); UpdateRequestBuilder builder = client.prepareUpdate(index, type, id) .setScript(getScript(data, map)) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); if (esRequestMapping.isUpsert()) { builder.setUpsert(getUpsert(data)).setScriptedUpsert(true); } return builder; } else { return client.prepareUpdate(index, type, id).setDoc(requestBodyMapper.map(data)) .setDocAsUpsert(esRequestMapping.isUpsert()) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); } } else { logger.warn("Updating doc by query, may affect performance"); return UpdateByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)) .script(getScript(data, data.getFields())); } default: throw new IllegalArgumentException("Unsupported row event type: " + data); } } }
ESRequestMapper implements Mapper<SyncData, Object> { @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override public Object map(SyncData data) { esQueryMapper.parseExtraQueryContext(data.getExtraQueryContext()); StandardEvaluationContext context = data.getContext(); String index = eval(indexExpr, context); String type = eval(typeExpr, context); String id = data.getId() == null ? null : data.getId().toString(); switch (data.getType()) { case WRITE: if (esRequestMapping.getNoUseIdForIndex()) { return client.prepareIndex(index, type).setSource(requestBodyMapper.map(data)); } return client.prepareIndex(index, type, id).setSource(requestBodyMapper.map(data)); case DELETE: logger.info("Deleting doc from Elasticsearch, may affect performance"); if (id != null) { return client.prepareDelete(index, type, id); } logger.warn("Deleting doc by query, may affect performance"); return DeleteByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)); case UPDATE: if (id != null) { if (needScript(data)) { HashMap<String, Object> map = requestBodyMapper.map(data); UpdateRequestBuilder builder = client.prepareUpdate(index, type, id) .setScript(getScript(data, map)) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); if (esRequestMapping.isUpsert()) { builder.setUpsert(getUpsert(data)).setScriptedUpsert(true); } return builder; } else { return client.prepareUpdate(index, type, id).setDoc(requestBodyMapper.map(data)) .setDocAsUpsert(esRequestMapping.isUpsert()) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); } } else { logger.warn("Updating doc by query, may affect performance"); return UpdateByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)) .script(getScript(data, data.getFields())); } default: throw new IllegalArgumentException("Unsupported row event type: " + data); } } ESRequestMapper(AbstractClient client, ESRequestMapping esRequestMapping); }
ESRequestMapper implements Mapper<SyncData, Object> { @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override public Object map(SyncData data) { esQueryMapper.parseExtraQueryContext(data.getExtraQueryContext()); StandardEvaluationContext context = data.getContext(); String index = eval(indexExpr, context); String type = eval(typeExpr, context); String id = data.getId() == null ? null : data.getId().toString(); switch (data.getType()) { case WRITE: if (esRequestMapping.getNoUseIdForIndex()) { return client.prepareIndex(index, type).setSource(requestBodyMapper.map(data)); } return client.prepareIndex(index, type, id).setSource(requestBodyMapper.map(data)); case DELETE: logger.info("Deleting doc from Elasticsearch, may affect performance"); if (id != null) { return client.prepareDelete(index, type, id); } logger.warn("Deleting doc by query, may affect performance"); return DeleteByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)); case UPDATE: if (id != null) { if (needScript(data)) { HashMap<String, Object> map = requestBodyMapper.map(data); UpdateRequestBuilder builder = client.prepareUpdate(index, type, id) .setScript(getScript(data, map)) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); if (esRequestMapping.isUpsert()) { builder.setUpsert(getUpsert(data)).setScriptedUpsert(true); } return builder; } else { return client.prepareUpdate(index, type, id).setDoc(requestBodyMapper.map(data)) .setDocAsUpsert(esRequestMapping.isUpsert()) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); } } else { logger.warn("Updating doc by query, may affect performance"); return UpdateByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)) .script(getScript(data, data.getFields())); } default: throw new IllegalArgumentException("Unsupported row event type: " + data); } } ESRequestMapper(AbstractClient client, ESRequestMapping esRequestMapping); @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override Object map(SyncData data); }
ESRequestMapper implements Mapper<SyncData, Object> { @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override public Object map(SyncData data) { esQueryMapper.parseExtraQueryContext(data.getExtraQueryContext()); StandardEvaluationContext context = data.getContext(); String index = eval(indexExpr, context); String type = eval(typeExpr, context); String id = data.getId() == null ? null : data.getId().toString(); switch (data.getType()) { case WRITE: if (esRequestMapping.getNoUseIdForIndex()) { return client.prepareIndex(index, type).setSource(requestBodyMapper.map(data)); } return client.prepareIndex(index, type, id).setSource(requestBodyMapper.map(data)); case DELETE: logger.info("Deleting doc from Elasticsearch, may affect performance"); if (id != null) { return client.prepareDelete(index, type, id); } logger.warn("Deleting doc by query, may affect performance"); return DeleteByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)); case UPDATE: if (id != null) { if (needScript(data)) { HashMap<String, Object> map = requestBodyMapper.map(data); UpdateRequestBuilder builder = client.prepareUpdate(index, type, id) .setScript(getScript(data, map)) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); if (esRequestMapping.isUpsert()) { builder.setUpsert(getUpsert(data)).setScriptedUpsert(true); } return builder; } else { return client.prepareUpdate(index, type, id).setDoc(requestBodyMapper.map(data)) .setDocAsUpsert(esRequestMapping.isUpsert()) .setRetryOnConflict(esRequestMapping.getRetryOnUpdateConflict()); } } else { logger.warn("Updating doc by query, may affect performance"); return UpdateByQueryAction.INSTANCE.newRequestBuilder(client) .source(index) .filter(getFilter(data)) .script(getScript(data, data.getFields())); } default: throw new IllegalArgumentException("Unsupported row event type: " + data); } } ESRequestMapper(AbstractClient client, ESRequestMapping esRequestMapping); @ThreadSafe(safe = {SpelExpressionParser.class, ESRequestMapping.class, TransportClient.class}) @Override Object map(SyncData data);  }