Skip to content

Commit a6f2fcd

Browse files
cgivrecgivre
committed
Updated to reflect DRILL-8248
1 parent 96d22f5 commit a6f2fcd

File tree

3 files changed

+120
-36
lines changed

3 files changed

+120
-36
lines changed

‎exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConvertFrom.java

Lines changed: 51 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
import org.apache.drill.exec.expr.annotations.Param;
2929
import org.apache.drill.exec.expr.annotations.Workspace;
3030
import org.apache.drill.exec.expr.holders.NullableVarBinaryHolder;
31-
import org.apache.drill.exec.expr.holders.VarCharHolder;
31+
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
3232
import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
3333
import org.apache.drill.exec.server.options.OptionManager;
3434
import org.apache.drill.exec.vector.complex.writer.BaseWriter;
@@ -45,23 +45,24 @@ public static class ConvertFromJsonNullableInput implements DrillSimpleFunc {
4545
@Param
4646
NullableVarBinaryHolder in;
4747

48-
@Workspace
49-
org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder jsonLoaderBuilder;
48+
@Output // TODO Remove in future work
49+
BaseWriter.ComplexWriter writer;
5050

5151
@Inject
5252
OptionManager options;
5353

5454
@Inject
55-
ResultSetLoader loader;
55+
ResultSetLoader rsLoader;
5656

57-
@Output // TODO Remove in future work
58-
BaseWriter.ComplexWriter writer;
57+
@Workspace
58+
org.apache.drill.exec.store.easy.json.loader.SingleElementIterator<java.io.InputStream> stream;
59+
60+
@Workspace
61+
org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl jsonLoader;
5962

6063
@Override
6164
public void setup() {
62-
jsonLoaderBuilder = new org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder()
63-
.resultSetLoader(loader)
64-
.standardOptions(options);
65+
rsLoader.startBatch();
6566
}
6667

6768
@Override
@@ -71,41 +72,57 @@ public void eval() {
7172
return;
7273
}
7374

75+
java.io.InputStream inputStream = org.apache.drill.exec.vector.complex.fn.DrillBufInputStream.getStream(in.start, in.end, in.buffer);
76+
7477
try {
75-
jsonLoaderBuilder.fromStream(in.start, in.end, in.buffer);
76-
org.apache.drill.exec.store.easy.json.loader.JsonLoader jsonLoader = jsonLoaderBuilder.build();
77-
loader.startBatch();
78-
jsonLoader.readBatch();
78+
stream.setValue(inputStream);
79+
80+
if (jsonLoader == null) {
81+
jsonLoader = org.apache.drill.exec.expr.fn.impl.conv.JsonConverterUtils.createJsonLoader(rsLoader, options, stream);
82+
}
83+
84+
org.apache.drill.exec.physical.resultSet.RowSetLoader rowWriter = rsLoader.writer();
85+
rowWriter.start();
86+
if (jsonLoader.parser().next()) {
87+
rowWriter.save();
88+
}
89+
inputStream.close();
90+
7991
} catch (Exception e) {
80-
throw new org.apache.drill.common.exceptions.DrillRuntimeException("Error while converting from JSON. ", e);
92+
throw org.apache.drill.common.exceptions.UserException.dataReadError(e)
93+
.message("Error while reading JSON. ")
94+
.addContext(e.getMessage())
95+
.build();
8196
}
8297
}
8398
}
8499

85100
@FunctionTemplate(names = {"convert_fromJSON", "convertFromJson", "convert_from_json"},
86-
scope = FunctionScope.SIMPLE)
101+
scope = FunctionScope.SIMPLE, nulls = NullHandling.INTERNAL)
87102
public static class ConvertFromJsonVarcharInput implements DrillSimpleFunc {
88103

89104
@Param
90-
VarCharHolder in;
105+
NullableVarCharHolder in;
91106

92107
@Output // TODO Remove in future work
93108
ComplexWriter writer;
94109

95110
@Workspace
96-
org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder jsonLoaderBuilder;
111+
org.apache.drill.exec.store.easy.json.loader.SingleElementIterator<java.io.InputStream> stream;
97112

98113
@Inject
99114
OptionManager options;
100115

101116
@Inject
102-
ResultSetLoader loader;
117+
ResultSetLoader rsLoader;
118+
119+
@Workspace
120+
org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl jsonLoader;
121+
103122

104123
@Override
105124
public void setup() {
106-
jsonLoaderBuilder = new org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder()
107-
.resultSetLoader(loader)
108-
.standardOptions(options);
125+
rsLoader.startBatch();
109126
}
110127

111128
@Override
@@ -118,12 +135,20 @@ public void eval() {
118135
}
119136

120137
try {
121-
jsonLoaderBuilder.fromString(jsonString);
122-
org.apache.drill.exec.store.easy.json.loader.JsonLoader jsonLoader = jsonLoaderBuilder.build();
123-
loader.startBatch();
124-
jsonLoader.readBatch();
138+
stream.setValue(org.apache.drill.exec.expr.fn.impl.conv.JsonConverterUtils.convertStringToInputStream(jsonString));
139+
if (jsonLoader == null) {
140+
jsonLoader = org.apache.drill.exec.expr.fn.impl.conv.JsonConverterUtils.createJsonLoader(rsLoader, options, stream);
141+
}
142+
org.apache.drill.exec.physical.resultSet.RowSetLoader rowWriter = rsLoader.writer();
143+
rowWriter.start();
144+
if (jsonLoader.parser().next()) {
145+
rowWriter.save();
146+
}
125147
} catch (Exception e) {
126-
throw new org.apache.drill.common.exceptions.DrillRuntimeException("Error while converting from JSON. ", e);
148+
throw org.apache.drill.common.exceptions.UserException.dataReadError(e)
149+
.message("Error while reading JSON. ")
150+
.addContext(e.getMessage())
151+
.build();
127152
}
128153
}
129154
}

‎exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConverterUtils.java

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.drill.exec.expr.fn.impl.conv;
20+
21+
22+
import org.apache.commons.io.IOUtils;
23+
import org.apache.drill.common.exceptions.UserException;
24+
import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
25+
import org.apache.drill.exec.server.options.OptionManager;
26+
import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl;
27+
import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder;
28+
import org.apache.drill.exec.store.easy.json.loader.SingleElementIterator;
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
31+
32+
import java.io.IOException;
33+
import java.io.InputStream;
34+
import java.nio.charset.Charset;
35+
36+
public class JsonConverterUtils {
37+
38+
private static final Logger logger = LoggerFactory.getLogger(JsonConverterUtils.class);
39+
40+
public static InputStream convertStringToInputStream(String input) {
41+
try (InputStream stream = IOUtils.toInputStream(input, Charset.defaultCharset())) {
42+
return stream;
43+
} catch (IOException e) {
44+
throw UserException.dataReadError(e)
45+
.message("Unable to read JSON string")
46+
.build(logger);
47+
}
48+
}
49+
50+
/**
51+
* Creates a {@link JsonLoaderImpl} for use in JSON conversion UDFs.
52+
* @param rsLoader The {@link ResultSetLoader} used in the UDF
53+
* @param options The {@link OptionManager} used in the UDF. This is used to extract the global JSON options
54+
* @param stream An input stream containing the input JSON data
55+
* @return A {@link JsonLoaderImpl} for use in the UDF.
56+
*/
57+
public static JsonLoaderImpl createJsonLoader(ResultSetLoader rsLoader,
58+
OptionManager options,
59+
SingleElementIterator<InputStream> stream) {
60+
// Add JSON configuration from Storage plugin, if present.
61+
JsonLoaderBuilder jsonLoaderBuilder = new JsonLoaderBuilder()
62+
.resultSetLoader(rsLoader)
63+
.standardOptions(options)
64+
.fromStream(() -> stream);
65+
66+
return (JsonLoaderImpl) jsonLoaderBuilder.build();
67+
}
68+
69+
}

‎exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/json/loader/JsonLoaderImpl.java

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -189,16 +189,6 @@ public JsonLoaderBuilder fromStream(InputStream... stream) {
189189
return this;
190190
}
191191

192-
public JsonLoaderBuilder fromStream(int start, int end, DrillBuf buf) {
193-
this.streams = Collections.singletonList(DrillBufInputStream.getStream(start, end, buf));
194-
return this;
195-
}
196-
197-
public JsonLoaderBuilder fromString(String jsonString) {
198-
this.streams = Collections.singletonList(IOUtils.toInputStream(jsonString, Charset.defaultCharset()));
199-
return this;
200-
}
201-
202192
public JsonLoaderBuilder fromStream(Iterable<InputStream> streams) {
203193
this.streams = streams;
204194
return this;

0 commit comments

Comments
 (0)