Introduce _sql/translate endpoint
Builds on RestSqlAction and in fact, extends SqlAction to keep up with future request settings. Original commit: elastic/x-pack-elasticsearch@7bbef4bdff
This commit is contained in:
parent
3e6714e205
commit
6cc3c067b7
|
@ -67,6 +67,7 @@ import org.elasticsearch.xpack.security.user.XPackSecurityUser;
|
|||
import org.elasticsearch.xpack.security.user.XPackUser;
|
||||
import org.elasticsearch.xpack.sql.plugin.jdbc.action.JdbcAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateAction;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
|
@ -489,7 +490,8 @@ public class AuthorizationService extends AbstractComponent {
|
|||
*/
|
||||
private static boolean isDelayedIndicesAction(String action) {
|
||||
return action.equals(SqlAction.NAME) ||
|
||||
action.equals(JdbcAction.NAME);
|
||||
action.equals(JdbcAction.NAME) ||
|
||||
action.equals(SqlTranslateAction.NAME);
|
||||
}
|
||||
|
||||
private static boolean isTranslatedToBulkAction(String action) {
|
||||
|
|
|
@ -12,14 +12,12 @@ import org.elasticsearch.common.unit.TimeValue;
|
|||
import org.elasticsearch.license.AbstractLicensesIntegrationTestCase;
|
||||
import org.elasticsearch.license.License;
|
||||
import org.elasticsearch.license.License.OperationMode;
|
||||
import org.elasticsearch.xpack.sql.jdbc.net.protocol.MetaTableRequest;
|
||||
import org.elasticsearch.xpack.sql.jdbc.net.protocol.MetaTableResponse;
|
||||
import org.elasticsearch.xpack.sql.plugin.jdbc.action.JdbcAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.jdbc.action.JdbcResponse;
|
||||
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
||||
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlResponse;
|
||||
import org.elasticsearch.xpack.sql.protocol.shared.Request;
|
||||
import org.elasticsearch.xpack.sql.protocol.shared.Response;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateResponse;
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Before;
|
||||
|
||||
|
@ -29,7 +27,6 @@ import static org.elasticsearch.license.XPackLicenseStateTests.randomBasicStanda
|
|||
import static org.elasticsearch.license.XPackLicenseStateTests.randomTrialBasicStandardGoldOrPlatinumMode;
|
||||
import static org.elasticsearch.license.XPackLicenseStateTests.randomTrialOrPlatinumMode;
|
||||
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
|
||||
import static org.hamcrest.Matchers.instanceOf;
|
||||
import static org.hamcrest.core.IsEqual.equalTo;
|
||||
|
||||
public class SqlLicenseIT extends AbstractLicensesIntegrationTestCase {
|
||||
|
@ -99,20 +96,20 @@ public class SqlLicenseIT extends AbstractLicensesIntegrationTestCase {
|
|||
assertThat(response.size(), Matchers.equalTo(2L));
|
||||
}
|
||||
|
||||
public void testJdbcActionLicense() throws Exception {
|
||||
public void testSqlTranslateActionLicense() throws Exception {
|
||||
setupTestIndex();
|
||||
disableJdbcLicensing();
|
||||
|
||||
Request request = new MetaTableRequest("test");
|
||||
disableSqlLicensing();
|
||||
|
||||
ElasticsearchSecurityException e = expectThrows(ElasticsearchSecurityException.class,
|
||||
() -> client().prepareExecute(JdbcAction.INSTANCE).request(request).get());
|
||||
assertThat(e.getMessage(), equalTo("current license is non-compliant for [jdbc]"));
|
||||
() -> client().prepareExecute(SqlTranslateAction.INSTANCE).query("SELECT * FROM test").get());
|
||||
assertThat(e.getMessage(), equalTo("current license is non-compliant for [sql]"));
|
||||
enableSqlLicensing();
|
||||
|
||||
enableJdbcLicensing();
|
||||
JdbcResponse jdbcResponse = client().prepareExecute(JdbcAction.INSTANCE).request(request).get();
|
||||
Response response = jdbcResponse.response(request);
|
||||
assertThat(response, instanceOf(MetaTableResponse.class));
|
||||
SqlTranslateResponse response = client().prepareExecute(SqlTranslateAction.INSTANCE).query("SELECT * FROM test").get();
|
||||
SearchSourceBuilder source = response.source();
|
||||
assertThat(source.docValueFields(), Matchers.contains("count"));
|
||||
FetchSourceContext fetchSource = source.fetchSource();
|
||||
assertThat(fetchSource.includes(), Matchers.arrayContaining("data"));
|
||||
}
|
||||
|
||||
// TODO test SqlGetIndicesAction. Skipping for now because of lack of serialization support.
|
||||
|
|
|
@ -0,0 +1,42 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql;
|
||||
|
||||
import org.elasticsearch.action.index.IndexRequest;
|
||||
import org.elasticsearch.action.support.WriteRequest;
|
||||
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
||||
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
|
||||
import org.elasticsearch.search.sort.SortBuilders;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateResponse;
|
||||
|
||||
import static java.util.Collections.singletonList;
|
||||
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
|
||||
|
||||
public class SqlTranslateActionIT extends AbstractSqlIntegTestCase {
|
||||
|
||||
public void testSqlTranslateAction() throws Exception {
|
||||
assertAcked(client().admin().indices().prepareCreate("test").get());
|
||||
client().prepareBulk()
|
||||
.add(new IndexRequest("test", "doc", "1").source("data", "bar", "count", 42))
|
||||
.add(new IndexRequest("test", "doc", "2").source("data", "baz", "count", 43))
|
||||
.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
|
||||
.get();
|
||||
ensureYellow("test");
|
||||
|
||||
boolean columnOrder = randomBoolean();
|
||||
String columns = columnOrder ? "data, count" : "count, data";
|
||||
SqlTranslateResponse response = client().prepareExecute(SqlTranslateAction.INSTANCE)
|
||||
.query("SELECT " + columns + " FROM test ORDER BY count").get();
|
||||
SearchSourceBuilder source = response.source();
|
||||
FetchSourceContext fetch = source.fetchSource();
|
||||
assertEquals(true, fetch.fetchSource());
|
||||
assertArrayEquals(new String[] { "data" }, fetch.includes());
|
||||
assertEquals(singletonList("count"), source.docValueFields());
|
||||
assertEquals(singletonList(SortBuilders.fieldSort("count")), source.sorts());
|
||||
}
|
||||
}
|
||||
|
|
@ -8,11 +8,16 @@ package org.elasticsearch.xpack.sql.execution;
|
|||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
||||
import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.Catalog;
|
||||
import org.elasticsearch.xpack.sql.execution.search.SourceGenerator;
|
||||
import org.elasticsearch.xpack.sql.expression.function.DefaultFunctionRegistry;
|
||||
import org.elasticsearch.xpack.sql.expression.function.FunctionRegistry;
|
||||
import org.elasticsearch.xpack.sql.optimizer.Optimizer;
|
||||
import org.elasticsearch.xpack.sql.parser.SqlParser;
|
||||
import org.elasticsearch.xpack.sql.plan.physical.EsQueryExec;
|
||||
import org.elasticsearch.xpack.sql.plan.physical.PhysicalPlan;
|
||||
import org.elasticsearch.xpack.sql.planner.Planner;
|
||||
import org.elasticsearch.xpack.sql.plugin.SqlGetIndicesAction;
|
||||
import org.elasticsearch.xpack.sql.session.Cursor;
|
||||
|
@ -59,6 +64,18 @@ public class PlanExecutor {
|
|||
functionRegistry, optimizer, planner);
|
||||
}
|
||||
|
||||
|
||||
public SearchSourceBuilder searchSource(String sql, SqlSettings settings) {
|
||||
PhysicalPlan executable = newSession(settings).executable(sql);
|
||||
if (executable instanceof EsQueryExec) {
|
||||
EsQueryExec e = (EsQueryExec) executable;
|
||||
return SourceGenerator.sourceBuilder(e.queryContainer(), settings.pageSize());
|
||||
}
|
||||
else {
|
||||
throw new SqlIllegalArgumentException("Cannot generate a query DSL for %s", sql);
|
||||
}
|
||||
}
|
||||
|
||||
public void sql(String sql, ActionListener<RowSetCursor> listener) {
|
||||
sql(SqlSettings.EMPTY, sql, listener);
|
||||
}
|
||||
|
|
|
@ -78,7 +78,7 @@ public class Scroller {
|
|||
|
||||
public void scroll(Schema schema, QueryContainer query, String index, ActionListener<RowSetCursor> listener) {
|
||||
// prepare the request
|
||||
SearchSourceBuilder sourceBuilder = SourceGenerator.sourceBuilder(query);
|
||||
SearchSourceBuilder sourceBuilder = SourceGenerator.sourceBuilder(query, size);
|
||||
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("About to execute query {} on {}", StringUtils.toString(sourceBuilder), index);
|
||||
|
@ -87,12 +87,6 @@ public class Scroller {
|
|||
SearchRequest search = client.prepareSearch(index).setSource(sourceBuilder).request();
|
||||
search.scroll(keepAlive).source().timeout(timeout);
|
||||
|
||||
// set the size only if it hasn't been specified (aggs only queries set the size to 0)
|
||||
if (search.source().size() == -1) {
|
||||
int sz = query.limit() > 0 ? Math.min(query.limit(), size) : size;
|
||||
search.source().size(sz);
|
||||
}
|
||||
|
||||
boolean isAggsOnly = query.isAggsOnly();
|
||||
|
||||
ScrollerActionListener l = isAggsOnly ? new AggsScrollActionListener(listener, client, timeout, schema, query) : new HandshakeScrollActionListener(listener, client, timeout, schema, query);
|
||||
|
@ -107,7 +101,7 @@ public class Scroller {
|
|||
|
||||
// dedicated scroll used for aggs-only/group-by results
|
||||
static class AggsScrollActionListener extends ScrollerActionListener {
|
||||
|
||||
|
||||
private final QueryContainer query;
|
||||
|
||||
AggsScrollActionListener(ActionListener<RowSetCursor> listener, Client client, TimeValue keepAlive, Schema schema, QueryContainer query) {
|
||||
|
@ -117,14 +111,14 @@ public class Scroller {
|
|||
|
||||
@Override
|
||||
protected RowSetCursor handleResponse(SearchResponse response) {
|
||||
|
||||
|
||||
final List<Object[]> extractedAggs = new ArrayList<>();
|
||||
AggValues aggValues = new AggValues(extractedAggs);
|
||||
List<Supplier<Object>> aggColumns = new ArrayList<>(query.columns().size());
|
||||
|
||||
// this method assumes the nested aggregation are all part of the same tree (the SQL group-by)
|
||||
int maxDepth = -1;
|
||||
|
||||
|
||||
List<ColumnReference> cols = query.columns();
|
||||
for (int index = 0; index < cols.size(); index++) {
|
||||
ColumnReference col = cols.get(index);
|
||||
|
@ -132,7 +126,7 @@ public class Scroller {
|
|||
|
||||
if (col instanceof ComputedRef) {
|
||||
ComputedRef pRef = (ComputedRef) col;
|
||||
|
||||
|
||||
Processor processor = pRef.processor().transformUp(a -> {
|
||||
Object[] value = extractAggValue(new AggRef(a.context()), response);
|
||||
extractedAggs.add(value);
|
||||
|
@ -179,14 +173,14 @@ public class Scroller {
|
|||
path = AggPath.bucketValueWithoutFormat(path);
|
||||
}
|
||||
Object value = getAggProperty(response.getAggregations(), path);
|
||||
|
||||
// // FIXME: this can be tabular in nature
|
||||
// if (ref instanceof MappedAggRef) {
|
||||
// Map<String, Object> map = (Map<String, Object>) value;
|
||||
|
||||
// // FIXME: this can be tabular in nature
|
||||
// if (ref instanceof MappedAggRef) {
|
||||
// Map<String, Object> map = (Map<String, Object>) value;
|
||||
// Object extractedValue = map.get(((MappedAggRef)
|
||||
// ref).fieldName());
|
||||
// }
|
||||
|
||||
// }
|
||||
|
||||
if (formattedKey) {
|
||||
List<? extends Bucket> buckets = ((MultiBucketsAggregation) value).getBuckets();
|
||||
arr = new Object[buckets.size()];
|
||||
|
@ -196,12 +190,12 @@ public class Scroller {
|
|||
} else {
|
||||
arr = value instanceof Object[] ? (Object[]) value : new Object[] { value };
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return arr;
|
||||
}
|
||||
}
|
||||
throw new SqlIllegalArgumentException("Unexpected non-agg/grouped column specified; %s", col.getClass());
|
||||
}
|
||||
}
|
||||
|
||||
private static Object getAggProperty(Aggregations aggs, String path) {
|
||||
List<String> list = AggregationPath.parse(path).getPathElementsAsStringList();
|
||||
|
@ -212,7 +206,7 @@ public class Scroller {
|
|||
}
|
||||
return agg.getProperty(list.subList(1, list.size()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// initial scroll used for parsing search hits (handles possible aggs)
|
||||
static class HandshakeScrollActionListener extends SearchHitsActionListener {
|
||||
|
|
|
@ -41,9 +41,7 @@ import java.util.LinkedHashSet;
|
|||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Set;
|
||||
|
||||
import static java.util.Collections.singletonList;
|
||||
import java.util.Set;import static java.util.Collections.singletonList;
|
||||
import static org.elasticsearch.search.sort.SortBuilders.fieldSort;
|
||||
import static org.elasticsearch.search.sort.SortBuilders.scriptSort;
|
||||
|
||||
|
@ -51,7 +49,7 @@ public abstract class SourceGenerator {
|
|||
|
||||
private static final List<String> NO_STORED_FIELD = singletonList(StoredFieldsContext._NONE_);
|
||||
|
||||
public static SearchSourceBuilder sourceBuilder(QueryContainer container) {
|
||||
public static SearchSourceBuilder sourceBuilder(QueryContainer container, Integer size) {
|
||||
SearchSourceBuilder source = new SearchSourceBuilder();
|
||||
// add the source
|
||||
if (container.query() != null) {
|
||||
|
@ -65,14 +63,14 @@ public abstract class SourceGenerator {
|
|||
|
||||
for (ColumnReference ref : container.columns()) {
|
||||
collectFields(ref, sourceFields, docFields, scriptFields);
|
||||
}
|
||||
}
|
||||
|
||||
if (!sourceFields.isEmpty()) {
|
||||
source.fetchSource(sourceFields.toArray(new String[sourceFields.size()]), null);
|
||||
}
|
||||
for (String field : docFields) {
|
||||
source.docValueField(field);
|
||||
}
|
||||
for (String field : docFields) {
|
||||
source.docValueField(field);
|
||||
}
|
||||
for (Entry<String, Script> entry : scriptFields.entrySet()) {
|
||||
source.scriptField(entry.getKey(), entry.getValue());
|
||||
}
|
||||
|
@ -92,6 +90,14 @@ public abstract class SourceGenerator {
|
|||
|
||||
optimize(container, source);
|
||||
|
||||
// set size
|
||||
if (size != null) {
|
||||
if (source.size() == -1) {
|
||||
int sz = container.limit() > 0 ? Math.min(container.limit(), size) : size;
|
||||
source.size(sz);
|
||||
}
|
||||
}
|
||||
|
||||
return source;
|
||||
}
|
||||
|
||||
|
|
|
@ -88,9 +88,7 @@ public class RestSqlCliAction extends BaseRestHandler {
|
|||
|
||||
private static Consumer<RestChannel> queryInit(Client client, QueryInitRequest request) {
|
||||
// TODO time zone support for CLI
|
||||
SqlRequest sqlRequest = new SqlRequest(request.query, SqlRequest.DEFAULT_TIME_ZONE, Cursor.EMPTY)
|
||||
.timeZone(DateTimeZone.forTimeZone(request.timeZone))
|
||||
.fetchSize(request.fetchSize);
|
||||
SqlRequest sqlRequest = new SqlRequest(request.query, DateTimeZone.forTimeZone(request.timeZone), request.fetchSize, Cursor.EMPTY);
|
||||
long start = System.nanoTime();
|
||||
return channel -> client.execute(SqlAction.INSTANCE, sqlRequest, toActionListener(channel, response -> {
|
||||
CliFormatter formatter = new CliFormatter(response);
|
||||
|
@ -108,7 +106,7 @@ public class RestSqlCliAction extends BaseRestHandler {
|
|||
} catch (IOException e) {
|
||||
throw new IllegalArgumentException("error reading the cursor");
|
||||
}
|
||||
SqlRequest sqlRequest = new SqlRequest("", SqlRequest.DEFAULT_TIME_ZONE, cursor);
|
||||
SqlRequest sqlRequest = new SqlRequest("", SqlRequest.DEFAULT_TIME_ZONE, -1, cursor);
|
||||
long start = System.nanoTime();
|
||||
return channel -> client.execute(SqlAction.INSTANCE, sqlRequest, toActionListener(channel, response -> {
|
||||
String data = formatter.formatWithoutHeader(response);
|
||||
|
|
|
@ -31,7 +31,9 @@ import org.elasticsearch.xpack.sql.plugin.jdbc.action.JdbcAction;
|
|||
import org.elasticsearch.xpack.sql.plugin.jdbc.action.JdbcHttpHandler;
|
||||
import org.elasticsearch.xpack.sql.plugin.jdbc.action.TransportJdbcAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.TransportSqlAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.TransportSqlTranslateAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.rest.RestSqlAction;
|
||||
import org.elasticsearch.xpack.sql.session.Cursor;
|
||||
|
||||
|
@ -86,6 +88,7 @@ public class SqlPlugin implements ActionPlugin {
|
|||
public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
|
||||
return Arrays.asList(new ActionHandler<>(SqlAction.INSTANCE, TransportSqlAction.class),
|
||||
new ActionHandler<>(JdbcAction.INSTANCE, TransportJdbcAction.class),
|
||||
new ActionHandler<>(SqlGetIndicesAction.INSTANCE, SqlGetIndicesAction.TransportAction.class));
|
||||
new ActionHandler<>(SqlGetIndicesAction.INSTANCE, SqlGetIndicesAction.TransportAction.class),
|
||||
new ActionHandler<>(SqlTranslateAction.INSTANCE, TransportSqlTranslateAction.class));
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,129 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
|
||||
|
||||
import org.elasticsearch.action.ActionRequest;
|
||||
import org.elasticsearch.action.CompositeIndicesRequest;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ObjectParser;
|
||||
import org.elasticsearch.xpack.sql.protocol.shared.AbstractQueryInitRequest;
|
||||
import org.joda.time.DateTimeZone;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Objects;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
public abstract class AbstractSqlRequest extends ActionRequest implements CompositeIndicesRequest {
|
||||
|
||||
public static final DateTimeZone DEFAULT_TIME_ZONE = DateTimeZone.UTC;
|
||||
public static final int DEFAULT_FETCH_SIZE = AbstractQueryInitRequest.DEFAULT_FETCH_SIZE;
|
||||
private String query = "";
|
||||
private DateTimeZone timeZone = DEFAULT_TIME_ZONE;
|
||||
private int fetchSize = DEFAULT_FETCH_SIZE;
|
||||
|
||||
public AbstractSqlRequest() {
|
||||
super();
|
||||
}
|
||||
|
||||
public AbstractSqlRequest(String query, DateTimeZone timeZone, int fetchSize) {
|
||||
this.query = query;
|
||||
this.timeZone = timeZone;
|
||||
this.fetchSize = fetchSize;
|
||||
}
|
||||
|
||||
public static <R extends AbstractSqlRequest> ObjectParser<R, Void> objectParser(Supplier<R> supplier) {
|
||||
ObjectParser<R, Void> parser = new ObjectParser<R, Void>("sql/query", supplier);
|
||||
|
||||
parser.declareString(AbstractSqlRequest::query, new ParseField("query"));
|
||||
parser.declareString((request, zoneId) -> request.timeZone(DateTimeZone.forID(zoneId)), new ParseField("time_zone"));
|
||||
parser.declareInt(AbstractSqlRequest::fetchSize, new ParseField("fetch_size"));
|
||||
|
||||
return parser;
|
||||
}
|
||||
|
||||
public String query() {
|
||||
return query;
|
||||
}
|
||||
|
||||
public AbstractSqlRequest query(String query) {
|
||||
if (query == null) {
|
||||
throw new IllegalArgumentException("query may not be null.");
|
||||
}
|
||||
this.query = query;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DateTimeZone timeZone() {
|
||||
return timeZone;
|
||||
}
|
||||
|
||||
public AbstractSqlRequest timeZone(DateTimeZone timeZone) {
|
||||
if (query == null) {
|
||||
throw new IllegalArgumentException("time zone may not be null.");
|
||||
}
|
||||
this.timeZone = timeZone;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hint about how many results to fetch at once.
|
||||
*/
|
||||
public int fetchSize() {
|
||||
return fetchSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hint about how many results to fetch at once.
|
||||
*/
|
||||
public AbstractSqlRequest fetchSize(int fetchSize) {
|
||||
if (fetchSize <= 0) {
|
||||
throw new IllegalArgumentException("fetch_size must be more than 0");
|
||||
}
|
||||
this.fetchSize = fetchSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
super.readFrom(in);
|
||||
query = in.readString();
|
||||
timeZone = DateTimeZone.forID(in.readString());
|
||||
fetchSize = in.readVInt();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
super.writeTo(out);
|
||||
out.writeString(query);
|
||||
out.writeString(timeZone.getID());
|
||||
out.writeVInt(fetchSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(query, timeZone, fetchSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (obj == null || getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
AbstractSqlRequest other = (AbstractSqlRequest) obj;
|
||||
return Objects.equals(query, other.query)
|
||||
&& Objects.equals(timeZone, other.timeZone)
|
||||
&& fetchSize == other.fetchSize;
|
||||
}
|
||||
}
|
|
@ -5,15 +5,12 @@
|
|||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.ActionRequest;
|
||||
import org.elasticsearch.action.ActionRequestValidationException;
|
||||
import org.elasticsearch.action.CompositeIndicesRequest;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ObjectParser;
|
||||
import org.elasticsearch.xpack.sql.protocol.shared.AbstractQueryInitRequest;
|
||||
import org.elasticsearch.xpack.sql.session.Cursor;
|
||||
import org.joda.time.DateTimeZone;
|
||||
|
||||
|
@ -22,74 +19,34 @@ import java.util.Objects;
|
|||
|
||||
import static org.elasticsearch.action.ValidateActions.addValidationError;
|
||||
|
||||
public class SqlRequest extends ActionRequest implements CompositeIndicesRequest {
|
||||
public class SqlRequest extends AbstractSqlRequest {
|
||||
|
||||
public static final ObjectParser<SqlRequest, Void> PARSER = objectParser(SqlRequest::new);
|
||||
|
||||
public static final ParseField CURSOR = new ParseField("cursor");
|
||||
public static final ObjectParser<SqlRequest, Void> PARSER = new ObjectParser<>("sql/query", SqlRequest::new);
|
||||
|
||||
static {
|
||||
PARSER.declareString(SqlRequest::query, new ParseField("query"));
|
||||
PARSER.declareString((request, zoneId) -> request.timeZone(DateTimeZone.forID(zoneId)), new ParseField("time_zone"));
|
||||
PARSER.declareInt(SqlRequest::fetchSize, new ParseField("fetch_size"));
|
||||
PARSER.declareString((request, nextPage) -> request.cursor(Cursor.decodeFromString(nextPage)), CURSOR);
|
||||
}
|
||||
|
||||
public static final DateTimeZone DEFAULT_TIME_ZONE = DateTimeZone.UTC;
|
||||
public static final int DEFAULT_FETCH_SIZE = AbstractQueryInitRequest.DEFAULT_FETCH_SIZE;
|
||||
|
||||
private String query = "";
|
||||
private DateTimeZone timeZone = DEFAULT_TIME_ZONE;
|
||||
private Cursor cursor = Cursor.EMPTY;
|
||||
private int fetchSize = DEFAULT_FETCH_SIZE;
|
||||
|
||||
public SqlRequest() {}
|
||||
|
||||
public SqlRequest(String query, DateTimeZone timeZone, Cursor cursor) {
|
||||
if (query == null) {
|
||||
throw new IllegalArgumentException("query must not be null");
|
||||
}
|
||||
if (timeZone == null) {
|
||||
throw new IllegalArgumentException("timeZone must not be null");
|
||||
}
|
||||
if (cursor == null) {
|
||||
throw new IllegalArgumentException("cursor must not be null");
|
||||
}
|
||||
this.query = query;
|
||||
this.timeZone = timeZone;
|
||||
public SqlRequest(String query, DateTimeZone timeZone, int fetchSize, Cursor cursor) {
|
||||
super(query, timeZone, fetchSize);
|
||||
this.cursor = cursor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ActionRequestValidationException validate() {
|
||||
ActionRequestValidationException validationException = null;
|
||||
if ((false == Strings.hasText(query)) && cursor == Cursor.EMPTY) {
|
||||
if ((false == Strings.hasText(query())) && cursor == Cursor.EMPTY) {
|
||||
validationException = addValidationError("one of [query] or [cursor] is required", validationException);
|
||||
}
|
||||
return validationException;
|
||||
}
|
||||
|
||||
public String query() {
|
||||
return query;
|
||||
}
|
||||
|
||||
public SqlRequest query(String query) {
|
||||
if (query == null) {
|
||||
throw new IllegalArgumentException("query may not be null.");
|
||||
}
|
||||
this.query = query;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DateTimeZone timeZone() {
|
||||
return timeZone;
|
||||
}
|
||||
|
||||
public SqlRequest timeZone(DateTimeZone timeZone) {
|
||||
if (query == null) {
|
||||
throw new IllegalArgumentException("time zone may not be null.");
|
||||
}
|
||||
this.timeZone = timeZone;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The key that must be sent back to SQL to access the next page of
|
||||
* results.
|
||||
|
@ -110,66 +67,31 @@ public class SqlRequest extends ActionRequest implements CompositeIndicesRequest
|
|||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hint about how many results to fetch at once.
|
||||
*/
|
||||
public int fetchSize() {
|
||||
return fetchSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hint about how many results to fetch at once.
|
||||
*/
|
||||
public SqlRequest fetchSize(int fetchSize) {
|
||||
if (fetchSize <= 0) {
|
||||
throw new IllegalArgumentException("fetch_size must be more than 0");
|
||||
}
|
||||
this.fetchSize = fetchSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
super.readFrom(in);
|
||||
query = in.readString();
|
||||
timeZone = DateTimeZone.forID(in.readString());
|
||||
cursor = in.readNamedWriteable(Cursor.class);
|
||||
fetchSize = in.readVInt();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
super.writeTo(out);
|
||||
out.writeString(query);
|
||||
out.writeString(timeZone.getID());
|
||||
out.writeNamedWriteable(cursor);
|
||||
out.writeVInt(fetchSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(query, timeZone, cursor);
|
||||
return Objects.hash(super.hashCode(), cursor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (obj == null || getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
SqlRequest other = (SqlRequest) obj;
|
||||
return Objects.equals(query, other.query)
|
||||
&& Objects.equals(timeZone, other.timeZone)
|
||||
&& Objects.equals(cursor, other.cursor)
|
||||
&& fetchSize == other.fetchSize;
|
||||
return super.equals(obj) && Objects.equals(cursor, ((SqlRequest) obj).cursor);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDescription() {
|
||||
return "SQL [" + query + "]";
|
||||
return "SQL [" + query() + "]";
|
||||
}
|
||||
}
|
|
@ -10,16 +10,17 @@ import org.elasticsearch.client.ElasticsearchClient;
|
|||
import org.elasticsearch.xpack.sql.session.Cursor;
|
||||
import org.joda.time.DateTimeZone;
|
||||
|
||||
import static org.elasticsearch.xpack.sql.plugin.sql.action.SqlRequest.DEFAULT_TIME_ZONE;
|
||||
import static org.elasticsearch.xpack.sql.plugin.sql.action.AbstractSqlRequest.DEFAULT_FETCH_SIZE;
|
||||
import static org.elasticsearch.xpack.sql.plugin.sql.action.AbstractSqlRequest.DEFAULT_TIME_ZONE;
|
||||
|
||||
public class SqlRequestBuilder extends ActionRequestBuilder<SqlRequest, SqlResponse, SqlRequestBuilder> {
|
||||
|
||||
public SqlRequestBuilder(ElasticsearchClient client, SqlAction action) {
|
||||
this(client, action, "", DEFAULT_TIME_ZONE, Cursor.EMPTY);
|
||||
this(client, action, "", DEFAULT_TIME_ZONE, DEFAULT_FETCH_SIZE, Cursor.EMPTY);
|
||||
}
|
||||
|
||||
public SqlRequestBuilder(ElasticsearchClient client, SqlAction action, String query, DateTimeZone timeZone, Cursor nextPageInfo) {
|
||||
super(client, action, new SqlRequest(query, timeZone, nextPageInfo));
|
||||
public SqlRequestBuilder(ElasticsearchClient client, SqlAction action, String query, DateTimeZone timeZone, int fetchSize, Cursor nextPageInfo) {
|
||||
super(client, action, new SqlRequest(query, timeZone, fetchSize, nextPageInfo));
|
||||
}
|
||||
|
||||
public SqlRequestBuilder query(String query) {
|
||||
|
|
|
@ -225,4 +225,4 @@ public class SqlResponse extends ActionResponse implements ToXContentObject {
|
|||
return Strings.toString(this);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
|
@ -0,0 +1,29 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.Action;
|
||||
import org.elasticsearch.client.ElasticsearchClient;
|
||||
|
||||
public class SqlTranslateAction extends Action<SqlTranslateRequest, SqlTranslateResponse, SqlTranslateRequestBuilder> {
|
||||
|
||||
public static final SqlTranslateAction INSTANCE = new SqlTranslateAction();
|
||||
public static final String NAME = "indices:data/read/sql/translate";
|
||||
|
||||
private SqlTranslateAction() {
|
||||
super(NAME);
|
||||
}
|
||||
|
||||
@Override
|
||||
public SqlTranslateRequestBuilder newRequestBuilder(ElasticsearchClient client) {
|
||||
return new SqlTranslateRequestBuilder(client, this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public SqlTranslateResponse newResponse() {
|
||||
return new SqlTranslateResponse();
|
||||
}
|
||||
}
|
|
@ -0,0 +1,38 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.ActionRequestValidationException;
|
||||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.xcontent.ObjectParser;
|
||||
import org.joda.time.DateTimeZone;
|
||||
|
||||
import static org.elasticsearch.action.ValidateActions.addValidationError;
|
||||
|
||||
public class SqlTranslateRequest extends AbstractSqlRequest {
|
||||
|
||||
public static final ObjectParser<SqlTranslateRequest, Void> PARSER = objectParser(SqlTranslateRequest::new);
|
||||
|
||||
public SqlTranslateRequest() {}
|
||||
|
||||
public SqlTranslateRequest(String query, DateTimeZone timeZone, int fetchSize) {
|
||||
super(query, timeZone, fetchSize);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ActionRequestValidationException validate() {
|
||||
ActionRequestValidationException validationException = null;
|
||||
if ((false == Strings.hasText(query()))) {
|
||||
validationException = addValidationError("query is required", validationException);
|
||||
}
|
||||
return validationException;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDescription() {
|
||||
return "SQL Translate [" + query() + "]";
|
||||
}
|
||||
}
|
|
@ -0,0 +1,35 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.ActionRequestBuilder;
|
||||
import org.elasticsearch.client.ElasticsearchClient;
|
||||
import org.joda.time.DateTimeZone;
|
||||
|
||||
import static org.elasticsearch.xpack.sql.plugin.sql.action.AbstractSqlRequest.DEFAULT_FETCH_SIZE;
|
||||
import static org.elasticsearch.xpack.sql.plugin.sql.action.AbstractSqlRequest.DEFAULT_TIME_ZONE;
|
||||
|
||||
public class SqlTranslateRequestBuilder
|
||||
extends ActionRequestBuilder<SqlTranslateRequest, SqlTranslateResponse, SqlTranslateRequestBuilder> {
|
||||
|
||||
public SqlTranslateRequestBuilder(ElasticsearchClient client, SqlTranslateAction action) {
|
||||
this(client, action, null, DEFAULT_TIME_ZONE, DEFAULT_FETCH_SIZE);
|
||||
}
|
||||
|
||||
public SqlTranslateRequestBuilder(ElasticsearchClient client, SqlTranslateAction action, String query, DateTimeZone timeZone, int fetchSize) {
|
||||
super(client, action, new SqlTranslateRequest(query, timeZone, fetchSize));
|
||||
}
|
||||
|
||||
public SqlTranslateRequestBuilder query(String query) {
|
||||
request.query(query);
|
||||
return this;
|
||||
}
|
||||
|
||||
public SqlTranslateRequestBuilder timeZone(DateTimeZone timeZone) {
|
||||
request.timeZone(timeZone);
|
||||
return this;
|
||||
}
|
||||
}
|
|
@ -0,0 +1,66 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.ActionResponse;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Objects;
|
||||
|
||||
public class SqlTranslateResponse extends ActionResponse implements ToXContentObject {
|
||||
|
||||
private SearchSourceBuilder source;
|
||||
|
||||
public SqlTranslateResponse() {
|
||||
}
|
||||
|
||||
public SqlTranslateResponse(SearchSourceBuilder source) {
|
||||
this.source = source;
|
||||
}
|
||||
|
||||
public SearchSourceBuilder source() {
|
||||
return source;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
source = new SearchSourceBuilder(in);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
source.writeTo(out);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (obj == null || getClass() != obj.getClass()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
SqlTranslateResponse other = (SqlTranslateResponse) obj;
|
||||
return Objects.equals(source, other.source);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(source);
|
||||
}
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
return source.toXContent(builder, params);
|
||||
}
|
||||
}
|
|
@ -0,0 +1,48 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.support.ActionFilters;
|
||||
import org.elasticsearch.action.support.HandledTransportAction;
|
||||
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
|
||||
import org.elasticsearch.common.inject.Inject;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
||||
import org.elasticsearch.xpack.sql.plugin.SqlLicenseChecker;
|
||||
import org.elasticsearch.xpack.sql.session.SqlSettings;
|
||||
|
||||
public class TransportSqlTranslateAction extends HandledTransportAction<SqlTranslateRequest, SqlTranslateResponse> {
|
||||
|
||||
private final PlanExecutor planExecutor;
|
||||
private final SqlLicenseChecker sqlLicenseChecker;
|
||||
|
||||
@Inject
|
||||
public TransportSqlTranslateAction(Settings settings, ThreadPool threadPool,
|
||||
TransportService transportService, ActionFilters actionFilters,
|
||||
IndexNameExpressionResolver indexNameExpressionResolver,
|
||||
PlanExecutor planExecutor,
|
||||
SqlLicenseChecker sqlLicenseChecker) {
|
||||
super(settings, SqlTranslateAction.NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, SqlTranslateRequest::new);
|
||||
|
||||
this.planExecutor = planExecutor;
|
||||
this.sqlLicenseChecker = sqlLicenseChecker;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doExecute(SqlTranslateRequest request, ActionListener<SqlTranslateResponse> listener) {
|
||||
sqlLicenseChecker.checkIfSqlAllowed();
|
||||
String query = request.query();
|
||||
|
||||
SqlSettings sqlSettings = new SqlSettings(Settings.builder()
|
||||
.put(SqlSettings.PAGE_SIZE, request.fetchSize())
|
||||
.put(SqlSettings.TIMEZONE_ID, request.timeZone().getID()).build());
|
||||
|
||||
listener.onResponse(new SqlTranslateResponse(planExecutor.searchSource(query, sqlSettings)));
|
||||
}
|
||||
}
|
|
@ -0,0 +1,46 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.rest;
|
||||
|
||||
import org.elasticsearch.client.node.NodeClient;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.rest.BaseRestHandler;
|
||||
import org.elasticsearch.rest.RestController;
|
||||
import org.elasticsearch.rest.RestRequest;
|
||||
import org.elasticsearch.rest.action.RestToXContentListener;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateRequest;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlTranslateResponse;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import static org.elasticsearch.rest.RestRequest.Method.GET;
|
||||
import static org.elasticsearch.rest.RestRequest.Method.POST;
|
||||
|
||||
public class RestSqlTranslateAction extends BaseRestHandler {
|
||||
|
||||
public RestSqlTranslateAction(Settings settings, RestController controller) {
|
||||
super(settings);
|
||||
controller.registerHandler(GET, "/_sql/translate", this);
|
||||
controller.registerHandler(POST, "/_sql/translate", this);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
|
||||
SqlTranslateRequest sqlRequest;
|
||||
try (XContentParser parser = request.contentOrSourceParamParser()) {
|
||||
sqlRequest = SqlTranslateRequest.PARSER.apply(parser, null);
|
||||
}
|
||||
|
||||
return channel -> client.executeLocally(SqlTranslateAction.INSTANCE, sqlRequest, new RestToXContentListener<SqlTranslateResponse>(channel));
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return "sql_translate_action";
|
||||
}
|
||||
}
|
|
@ -6,6 +6,9 @@
|
|||
package org.elasticsearch.xpack.sql.querydsl.container;
|
||||
|
||||
import org.elasticsearch.common.collect.Tuple;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.json.JsonXContent;
|
||||
import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
|
||||
import org.elasticsearch.xpack.sql.execution.search.SourceGenerator;
|
||||
import org.elasticsearch.xpack.sql.expression.Attribute;
|
||||
|
@ -23,8 +26,8 @@ import org.elasticsearch.xpack.sql.querydsl.query.AndQuery;
|
|||
import org.elasticsearch.xpack.sql.querydsl.query.MatchAll;
|
||||
import org.elasticsearch.xpack.sql.querydsl.query.NestedQuery;
|
||||
import org.elasticsearch.xpack.sql.querydsl.query.Query;
|
||||
import org.elasticsearch.xpack.sql.util.StringUtils;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
|
@ -96,7 +99,7 @@ public class QueryContainer {
|
|||
// check field references
|
||||
if (((ComputedRef) ref).processor().anyMatch(p -> p instanceof ReferenceInput && ((ReferenceInput) p).context() instanceof FieldReference)) {
|
||||
onlyAggs = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ref instanceof FieldReference) {
|
||||
onlyAggs = false;
|
||||
|
@ -243,7 +246,7 @@ public class QueryContainer {
|
|||
if (proc == null) {
|
||||
if (name instanceof ScalarFunctionAttribute) {
|
||||
sfa = (ScalarFunctionAttribute) name;
|
||||
}
|
||||
}
|
||||
proc = sfa.processorDef();
|
||||
}
|
||||
AtomicReference<QueryContainer> containerRef = new AtomicReference<QueryContainer>(this);
|
||||
|
@ -273,7 +276,7 @@ public class QueryContainer {
|
|||
private Tuple<QueryContainer, ColumnReference> toReference(Attribute attr) {
|
||||
if (attr instanceof RootFieldAttribute) {
|
||||
return new Tuple<>(this, fieldRef((RootFieldAttribute) attr));
|
||||
}
|
||||
}
|
||||
if (attr instanceof NestedFieldAttribute) {
|
||||
return nestedFieldRef((NestedFieldAttribute) attr);
|
||||
}
|
||||
|
@ -355,7 +358,12 @@ public class QueryContainer {
|
|||
|
||||
@Override
|
||||
public String toString() {
|
||||
// annoying cycle (QC depends on SC which depends on QC)
|
||||
return StringUtils.toString(SourceGenerator.sourceBuilder(this));
|
||||
try (XContentBuilder builder = JsonXContent.contentBuilder()) {
|
||||
builder.humanReadable(true).prettyPrint();
|
||||
SourceGenerator.sourceBuilder(this, null).toXContent(builder, ToXContent.EMPTY_PARAMS);
|
||||
return builder.string();
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException("error rendering", e);
|
||||
}
|
||||
}
|
||||
}
|
|
@ -21,8 +21,7 @@ public class SqlRequestTests extends AbstractStreamableTestCase<SqlRequest> {
|
|||
|
||||
@Override
|
||||
protected SqlRequest createTestInstance() {
|
||||
return new SqlRequest(randomAlphaOfLength(10), randomDateTimeZone(), randomCursor())
|
||||
.fetchSize(between(1, Integer.MAX_VALUE));
|
||||
return new SqlRequest(randomAlphaOfLength(10), randomDateTimeZone(), between(1, Integer.MAX_VALUE), randomCursor());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
@ -34,13 +33,13 @@ public class SqlRequestTests extends AbstractStreamableTestCase<SqlRequest> {
|
|||
@SuppressWarnings("unchecked")
|
||||
protected MutateFunction<SqlRequest> getMutateFunction() {
|
||||
return randomFrom(
|
||||
request -> getCopyFunction().copy(request)
|
||||
.query(randomValueOtherThan(request.query(), () -> randomAlphaOfLength(5))),
|
||||
request -> getCopyFunction().copy(request)
|
||||
.timeZone(randomValueOtherThan(request.timeZone(), ESTestCase::randomDateTimeZone)),
|
||||
request -> getCopyFunction().copy(request)
|
||||
.cursor(randomValueOtherThan(request.cursor(), SqlResponseTests::randomCursor)),
|
||||
request -> getCopyFunction().copy(request)
|
||||
request -> (SqlRequest) getCopyFunction().copy(request)
|
||||
.query(randomValueOtherThan(request.query(), () -> randomAlphaOfLength(5))),
|
||||
request -> (SqlRequest) getCopyFunction().copy(request)
|
||||
.timeZone(randomValueOtherThan(request.timeZone(), ESTestCase::randomDateTimeZone)),
|
||||
request -> (SqlRequest) getCopyFunction().copy(request)
|
||||
.fetchSize(randomValueOtherThan(request.fetchSize(), () -> between(1, Integer.MAX_VALUE))));
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,35 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.test.AbstractStreamableTestCase;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.test.EqualsHashCodeTestUtils.MutateFunction;
|
||||
|
||||
public class SqlTranslateRequestTests extends AbstractStreamableTestCase<SqlTranslateRequest> {
|
||||
|
||||
@Override
|
||||
protected SqlTranslateRequest createTestInstance() {
|
||||
return new SqlTranslateRequest(randomAlphaOfLength(10), randomDateTimeZone(), between(1, Integer.MAX_VALUE));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SqlTranslateRequest createBlankInstance() {
|
||||
return new SqlTranslateRequest();
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
protected MutateFunction<SqlTranslateRequest> getMutateFunction() {
|
||||
return randomFrom(
|
||||
request -> (SqlTranslateRequest) getCopyFunction().copy(request)
|
||||
.query(randomValueOtherThan(request.query(), () -> randomAlphaOfLength(5))),
|
||||
request -> (SqlTranslateRequest) getCopyFunction().copy(request)
|
||||
.timeZone(randomValueOtherThan(request.timeZone(), ESTestCase::randomDateTimeZone)),
|
||||
request -> (SqlTranslateRequest) getCopyFunction().copy(request)
|
||||
.fetchSize(randomValueOtherThan(request.fetchSize(), () -> between(1, Integer.MAX_VALUE))));
|
||||
}
|
||||
}
|
|
@ -0,0 +1,40 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin.sql.action;
|
||||
|
||||
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
||||
import org.elasticsearch.test.AbstractStreamableTestCase;
|
||||
|
||||
public class SqlTranslateResponseTests extends AbstractStreamableTestCase<SqlTranslateResponse> {
|
||||
|
||||
@Override
|
||||
protected SqlTranslateResponse createTestInstance() {
|
||||
SearchSourceBuilder s = new SearchSourceBuilder();
|
||||
if (randomBoolean()) {
|
||||
long docValues = iterations(5, 10);
|
||||
for (int i = 0; i < docValues; i++) {
|
||||
s.docValueField(randomAlphaOfLength(10));
|
||||
}
|
||||
}
|
||||
|
||||
if (randomBoolean()) {
|
||||
long sourceFields = iterations(5, 10);
|
||||
for (int i = 0; i < sourceFields; i++) {
|
||||
s.storedField(randomAlphaOfLength(10));
|
||||
}
|
||||
}
|
||||
|
||||
s.fetchSource(randomBoolean()).from(randomInt(256)).explain(randomBoolean()).size(randomInt(256));
|
||||
|
||||
return new SqlTranslateResponse(s);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected SqlTranslateResponse createBlankInstance() {
|
||||
return new SqlTranslateResponse();
|
||||
}
|
||||
|
||||
}
|
Loading…
Reference in New Issue