Skip to content

Commit f335101

Browse files
Yury-FridlyandYury-Fridlyand
committed
First PoC.
Signed-off-by: Yury-Fridlyand <[email protected]>
1 parent 662a938 commit f335101

File tree

7 files changed

+204
-24
lines changed

7 files changed

+204
-24
lines changed

‎core/src/main/java/org/opensearch/sql/data/model/ExprTimeValue.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,21 @@ public LocalTime timeValue() {
5757
return time;
5858
}
5959

60+
@Override
61+
public LocalDate dateValue() {
62+
return LocalDate.now();
63+
}
64+
65+
@Override
66+
public LocalDateTime datetimeValue() {
67+
return LocalDateTime.of(dateValue(), timeValue());
68+
}
69+
70+
@Override
71+
public Instant timestampValue() {
72+
return ZonedDateTime.of(dateValue(), timeValue(), ExprTimestampValue.ZONE).toInstant();
73+
}
74+
6075
@Override
6176
public String toString() {
6277
return String.format("TIME '%s'", value());

‎core/src/main/java/org/opensearch/sql/data/model/ExprTimestampValue.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ public class ExprTimestampValue extends AbstractExprValue {
3030
/**
3131
* todo. only support UTC now.
3232
*/
33-
private static final ZoneId ZONE = ZoneId.of("UTC");
33+
public static final ZoneId ZONE = ZoneId.of("UTC");
3434

3535
private final Instant timestamp;
3636

‎opensearch/src/main/java/org/opensearch/sql/opensearch/data/value/OpenSearchExprValueFactory.java‎

Lines changed: 88 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -30,16 +30,22 @@
3030
import com.fasterxml.jackson.core.JsonProcessingException;
3131
import com.fasterxml.jackson.databind.ObjectMapper;
3232
import com.google.common.collect.ImmutableMap;
33+
import java.time.DateTimeException;
3334
import java.time.Instant;
35+
import java.time.LocalDate;
36+
import java.time.LocalDateTime;
37+
import java.time.LocalTime;
3438
import java.time.format.DateTimeParseException;
39+
import java.time.temporal.TemporalAccessor;
3540
import java.util.ArrayList;
3641
import java.util.LinkedHashMap;
3742
import java.util.List;
3843
import java.util.Map;
3944
import java.util.Optional;
40-
import java.util.function.Function;
45+
import java.util.function.BiFunction;
4146
import lombok.Getter;
4247
import lombok.Setter;
48+
import org.apache.logging.log4j.LogManager;
4349
import org.opensearch.common.time.DateFormatters;
4450
import org.opensearch.sql.data.model.ExprBooleanValue;
4551
import org.opensearch.sql.data.model.ExprByteValue;
@@ -61,6 +67,7 @@
6167
import org.opensearch.sql.opensearch.data.utils.Content;
6268
import org.opensearch.sql.opensearch.data.utils.ObjectContent;
6369
import org.opensearch.sql.opensearch.data.utils.OpenSearchJsonContent;
70+
import org.opensearch.sql.opensearch.mapping.MappingEntry;
6471
import org.opensearch.sql.opensearch.response.agg.OpenSearchAggregationResponseParser;
6572

6673
/**
@@ -73,6 +80,9 @@ public class OpenSearchExprValueFactory {
7380
@Setter
7481
private Map<String, ExprType> typeMapping;
7582

83+
84+
private Map<String, MappingEntry> typeMapping2;
85+
7686
@Getter
7787
@Setter
7888
private OpenSearchAggregationResponseParser parser;
@@ -81,26 +91,26 @@ public class OpenSearchExprValueFactory {
8191

8292
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
8393

84-
private final Map<ExprType, Function<Content, ExprValue>> typeActionMap =
85-
new ImmutableMap.Builder<ExprType, Function<Content, ExprValue>>()
86-
.put(INTEGER, c -> new ExprIntegerValue(c.intValue()))
87-
.put(LONG, c -> new ExprLongValue(c.longValue()))
88-
.put(SHORT, c -> new ExprShortValue(c.shortValue()))
89-
.put(BYTE, c -> new ExprByteValue(c.byteValue()))
90-
.put(FLOAT, c -> new ExprFloatValue(c.floatValue()))
91-
.put(DOUBLE, c -> new ExprDoubleValue(c.doubleValue()))
92-
.put(STRING, c -> new ExprStringValue(c.stringValue()))
93-
.put(BOOLEAN, c -> ExprBooleanValue.of(c.booleanValue()))
94+
private final Map<ExprType, BiFunction<Content, MappingEntry, ExprValue>> typeActionMap =
95+
new ImmutableMap.Builder<ExprType, BiFunction<Content, MappingEntry, ExprValue>>()
96+
.put(INTEGER, (c, m) -> new ExprIntegerValue(c.intValue()))
97+
.put(LONG, (c, m) -> new ExprLongValue(c.longValue()))
98+
.put(SHORT, (c, m) -> new ExprShortValue(c.shortValue()))
99+
.put(BYTE, (c, m) -> new ExprByteValue(c.byteValue()))
100+
.put(FLOAT, (c, m) -> new ExprFloatValue(c.floatValue()))
101+
.put(DOUBLE, (c, m) -> new ExprDoubleValue(c.doubleValue()))
102+
.put(STRING, (c, m) -> new ExprStringValue(c.stringValue()))
103+
.put(BOOLEAN, (c, m) -> ExprBooleanValue.of(c.booleanValue()))
94104
.put(TIMESTAMP, this::parseTimestamp)
95-
.put(DATE, c -> new ExprDateValue(parseTimestamp(c).dateValue().toString()))
96-
.put(TIME, c -> new ExprTimeValue(parseTimestamp(c).timeValue().toString()))
97-
.put(DATETIME, c -> new ExprDatetimeValue(parseTimestamp(c).datetimeValue()))
98-
.put(OPENSEARCH_TEXT, c -> new OpenSearchExprTextValue(c.stringValue()))
99-
.put(OPENSEARCH_TEXT_KEYWORD, c -> new OpenSearchExprTextKeywordValue(c.stringValue()))
100-
.put(OPENSEARCH_IP, c -> new OpenSearchExprIpValue(c.stringValue()))
101-
.put(OPENSEARCH_GEO_POINT, c -> new OpenSearchExprGeoPointValue(c.geoValue().getLeft(),
105+
.put(DATE, (c, m) -> new ExprDateValue(parseTimestamp(c, m).dateValue().toString()))
106+
.put(TIME, (c, m) -> new ExprTimeValue(parseTimestamp(c, m).timeValue().toString()))
107+
.put(DATETIME, (c, m) -> new ExprDatetimeValue(parseTimestamp(c, m).datetimeValue()))
108+
.put(OPENSEARCH_TEXT, (c, m) -> new OpenSearchExprTextValue(c.stringValue()))
109+
.put(OPENSEARCH_TEXT_KEYWORD, (c, m) -> new OpenSearchExprTextKeywordValue(c.stringValue()))
110+
.put(OPENSEARCH_IP, (c, m) -> new OpenSearchExprIpValue(c.stringValue()))
111+
.put(OPENSEARCH_GEO_POINT, (c, m) -> new OpenSearchExprGeoPointValue(c.geoValue().getLeft(),
102112
c.geoValue().getRight()))
103-
.put(OPENSEARCH_BINARY, c -> new OpenSearchExprBinaryValue(c.stringValue()))
113+
.put(OPENSEARCH_BINARY, (c, m) -> new OpenSearchExprBinaryValue(c.stringValue()))
104114
.build();
105115

106116
/**
@@ -111,6 +121,12 @@ public OpenSearchExprValueFactory(
111121
this.typeMapping = typeMapping;
112122
}
113123

124+
public OpenSearchExprValueFactory(Map<String, ExprType> typeMapping,
125+
Map<String, MappingEntry> typeMapping2) {
126+
this.typeMapping = typeMapping;
127+
this.typeMapping2 = typeMapping2;
128+
}
129+
114130
/**
115131
* The struct construction has the following assumption. 1. The field has OpenSearch Object
116132
* data type. https://www.elastic.co/guide/en/elasticsearch/reference/current/object.html 2. The
@@ -151,7 +167,7 @@ private ExprValue parse(Content content, String field, Optional<ExprType> fieldT
151167
return parseArray(content, field);
152168
} else {
153169
if (typeActionMap.containsKey(type)) {
154-
return typeActionMap.get(type).apply(content);
170+
return typeActionMap.get(type).apply(content, typeMapping2.getOrDefault(field, null));
155171
} else {
156172
throw new IllegalStateException(
157173
String.format(
@@ -188,11 +204,61 @@ private ExprValue constructTimestamp(String value) {
188204
}
189205
}
190206

191-
private ExprValue parseTimestamp(Content value) {
207+
// returns java.time.format.Parsed
208+
private TemporalAccessor parseTimestampString(String value, MappingEntry mapping) {
209+
if (mapping == null) {
210+
return null;
211+
}
212+
for (var formatter : mapping.getRegularFormatters()) {
213+
try {
214+
return formatter.parse(value);
215+
} catch (Exception ignored) {
216+
// nothing to do, try another format
217+
}
218+
}
219+
for (var formatter : mapping.getNamedFormatters()) {
220+
try {
221+
return formatter.parse(value);
222+
} catch (Exception ignored) {
223+
// nothing to do, try another format
224+
}
225+
}
226+
return null;
227+
}
228+
229+
private ExprValue parseTimestamp(Content value, MappingEntry mapping) {
192230
if (value.isNumber()) {
193231
return new ExprTimestampValue(Instant.ofEpochMilli(value.longValue()));
194232
} else if (value.isString()) {
195-
return constructTimestamp(value.stringValue());
233+
TemporalAccessor parsed = parseTimestampString(value.stringValue(), mapping);
234+
if (parsed == null) { // failed to parse or no formats given
235+
return constructTimestamp(value.stringValue());
236+
}
237+
try {
238+
return new ExprTimestampValue(Instant.from(parsed));
239+
} catch (DateTimeException ignored) {
240+
// nothing to do, try another type
241+
}
242+
// TODO return not ExprTimestampValue
243+
try {
244+
return new ExprTimestampValue(new ExprDateValue(LocalDate.from(parsed)).timestampValue());
245+
} catch (DateTimeException ignored) {
246+
// nothing to do, try another type
247+
}
248+
try {
249+
return new ExprTimestampValue(new ExprDatetimeValue(LocalDateTime.from(parsed)).timestampValue());
250+
} catch (DateTimeException ignored) {
251+
// nothing to do, try another type
252+
}
253+
try {
254+
return new ExprTimestampValue(new ExprTimeValue(LocalTime.from(parsed)).timestampValue());
255+
} catch (DateTimeException ignored) {
256+
// nothing to do, try another type
257+
}
258+
// TODO throw exception
259+
LogManager.getLogger(OpenSearchExprValueFactory.class).error(
260+
String.format("Can't recognize parsed value: %s, %s", parsed, parsed.getClass()));
261+
return new ExprStringValue(value.stringValue());
196262
} else {
197263
return new ExprTimestampValue((Instant) value.objectValue());
198264
}

‎opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/IndexMapping.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,14 @@ public class IndexMapping {
2626
/** Field mappings from field name to field type in OpenSearch date type system. */
2727
private final Map<String, String> fieldMappings;
2828

29+
public Map<String, MappingEntry> mapping2;
30+
2931
public IndexMapping(Map<String, String> fieldMappings) {
3032
this.fieldMappings = fieldMappings;
3133
}
3234

3335
public IndexMapping(MappingMetadata metaData) {
36+
this.mapping2 = flat2(metaData.getSourceAsMap());
3437
this.fieldMappings = flatMappings(metaData.getSourceAsMap());
3538
}
3639

@@ -65,6 +68,17 @@ public <T> Map<String, T> getAllFieldTypes(Function<String, T> transform) {
6568
.collect(Collectors.toMap(Map.Entry::getKey, e -> transform.apply(e.getValue())));
6669
}
6770

71+
@SuppressWarnings("unchecked")
72+
private Map<String, MappingEntry> flat2(Map<String, Object> indexMapping) {
73+
return ((Map<String, Object>)indexMapping.get("properties")).entrySet().stream()
74+
.collect(Collectors.toMap(e -> e.getKey(), e -> {
75+
Map<String, Object> mapping = (Map<String, Object>) e.getValue();
76+
return new MappingEntry((String) mapping.getOrDefault("type", "object"),
77+
(String) mapping.getOrDefault("format", null), null);
78+
}));
79+
}
80+
81+
6882
@SuppressWarnings("unchecked")
6983
private Map<String, String> flatMappings(Map<String, Object> indexMapping) {
7084
ImmutableMap.Builder<String, String> builder = new ImmutableMap.Builder<>();

‎opensearch/src/main/java/org/opensearch/sql/opensearch/mapping/MappingEntry.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
7+
package org.opensearch.sql.opensearch.mapping;
8+
9+
import java.time.format.DateTimeFormatter;
10+
import java.util.Arrays;
11+
import java.util.List;
12+
import java.util.Objects;
13+
import java.util.stream.Collectors;
14+
import lombok.AllArgsConstructor;
15+
import lombok.Getter;
16+
import lombok.Setter;
17+
import org.opensearch.common.time.DateFormatter;
18+
import org.opensearch.sql.data.type.ExprType;
19+
20+
@AllArgsConstructor
21+
public class MappingEntry {
22+
23+
@Getter
24+
private String fieldType;
25+
26+
@Getter
27+
private String formats;
28+
29+
@Getter
30+
@Setter
31+
private ExprType dataType;
32+
33+
public MappingEntry(String fieldType) {
34+
this(fieldType, null, null);
35+
}
36+
37+
public List<String> getFormatList() {
38+
if (formats == null || formats.isEmpty()) {
39+
return List.of();
40+
}
41+
return Arrays.stream(formats.split("\\|\\|")).map(String::trim).collect(Collectors.toList());
42+
}
43+
44+
public List<DateFormatter> getNamedFormatters() {
45+
return getFormatList().stream().filter(f -> {
46+
try {
47+
DateTimeFormatter.ofPattern(f);
48+
return false;
49+
} catch (Exception e) {
50+
return true;
51+
}
52+
})
53+
.map(DateFormatter::forPattern).collect(Collectors.toList());
54+
}
55+
56+
public List<DateTimeFormatter> getRegularFormatters() {
57+
return getFormatList().stream().map(f -> {
58+
try {
59+
return DateTimeFormatter.ofPattern(f);
60+
} catch (Exception e) {
61+
return null;
62+
}
63+
})
64+
.filter(Objects::nonNull).collect(Collectors.toList());
65+
}
66+
}

‎opensearch/src/main/java/org/opensearch/sql/opensearch/request/system/OpenSearchDescribeIndexRequest.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,17 @@
1717
import java.util.List;
1818
import java.util.Map;
1919
import java.util.stream.Collectors;
20+
21+
import org.apache.commons.lang3.tuple.Triple;
22+
import org.opensearch.common.collect.Tuple;
2023
import org.opensearch.sql.data.model.ExprTupleValue;
2124
import org.opensearch.sql.data.model.ExprValue;
2225
import org.opensearch.sql.data.type.ExprCoreType;
2326
import org.opensearch.sql.data.type.ExprType;
2427
import org.opensearch.sql.opensearch.client.OpenSearchClient;
2528
import org.opensearch.sql.opensearch.data.type.OpenSearchDataType;
2629
import org.opensearch.sql.opensearch.mapping.IndexMapping;
30+
import org.opensearch.sql.opensearch.mapping.MappingEntry;
2731
import org.opensearch.sql.opensearch.request.OpenSearchRequest;
2832

2933
/**
@@ -121,6 +125,20 @@ public Map<String, ExprType> getFieldTypes() {
121125
return fieldTypes;
122126
}
123127

128+
// TODO possible collision if two indices have fields with same names
129+
public Map<String, MappingEntry> getFieldTypes2() {
130+
Map<String, IndexMapping> indexMappings = client.getIndexMappings(indexName.getIndexNames());
131+
Map<String, MappingEntry> fieldTypes = new HashMap<>();
132+
133+
for (IndexMapping indexMapping : indexMappings.values()) {
134+
indexMapping.mapping2.forEach((key, value) ->
135+
value.setDataType(transformESTypeToExprType(value.getFieldType())));
136+
fieldTypes
137+
.putAll(indexMapping.mapping2);
138+
}
139+
return fieldTypes;
140+
}
141+
124142
/**
125143
* Get the minimum of the max result windows of the indices.
126144
*

‎opensearch/src/main/java/org/opensearch/sql/opensearch/storage/OpenSearchIndex.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,8 @@ public Integer getMaxResultWindow() {
104104
@Override
105105
public PhysicalPlan implement(LogicalPlan plan) {
106106
OpenSearchIndexScan indexScan = new OpenSearchIndexScan(client, settings, indexName,
107-
getMaxResultWindow(), new OpenSearchExprValueFactory(getFieldTypes()));
107+
getMaxResultWindow(), new OpenSearchExprValueFactory(getFieldTypes(),
108+
new OpenSearchDescribeIndexRequest(client, indexName).getFieldTypes2()));
108109

109110
/*
110111
* Visit logical plan with index scan as context so logical operators visited, such as

0 commit comments

Comments
 (0)