You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by GitBox <gi...@apache.org> on 2022/05/29 02:43:57 UTC
[GitHub] [drill] cgivre opened a new pull request, #2567: DRILL-8239: Convert JSON UDF to EVF
cgivre opened a new pull request, #2567:
URL: https://github.com/apache/drill/pull/2567
# [DRILL-8239](https://issues.apache.org/jira/browse/DRILL-8239): Convert JSON UDF to EVF
## Description
This PR updates the `convert_fromJSON` UDFs to use the latest JSON reader.
## Documentation
No user facing changes.
## Testing
Ran unit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "shfshihuafeng (via GitHub)" <gi...@apache.org>.
shfshihuafeng commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3771229111
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
In my opinion, the loop body calling setTargetRowCount in ProjectRecordBatch.setValueCount() is dead code.
Two cases exist:
With a ComplexWriter function (convert_fromJSON, split, etc.): addComplexField() is called → complexWriters is non-null( initComplexWriters()->projectBatch.complexWriters = new ArrayList<>();) → complexWriters != null is true → enters the first branch, else if is short‑circuited.
Without any ComplexWriter function (e.g. select *): complexWriters stays null → complexWriters != null is false → the else if condition is evaluated. But addLoader() is also not called → rsLoaders is null → !CollectionUtils.isEmpty(rsLoaders) is false → the loop body is not entered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [drill] cgivre commented on a diff in pull request #2567: [WIP] DRILL-8239: Convert JSON UDF to EVF
Posted by GitBox <gi...@apache.org>.
cgivre commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r891318837
##########
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConvertFrom.java:
##########
@@ -32,37 +30,45 @@
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
import org.apache.drill.exec.expr.holders.VarBinaryHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;
public class JsonConvertFrom {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JsonConvertFrom.class);
-
private JsonConvertFrom() {
}
@FunctionTemplate(name = "convert_fromJSON", scope = FunctionScope.SIMPLE, isRandom = true)
public static class ConvertFromJson implements DrillSimpleFunc {
@Param VarBinaryHolder in;
- @Inject DrillBuf buffer;
- @Workspace org.apache.drill.exec.vector.complex.fn.JsonReader jsonReader;
+ @Inject
+ ResultSetLoader loader;
+ @Workspace
+ org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder jsonLoaderBuilder;
+
+ @Inject
+ OptionManager options;
@Output ComplexWriter writer;
@Override
public void setup() {
- jsonReader = new org.apache.drill.exec.vector.complex.fn.JsonReader.Builder(buffer)
- .defaultSchemaPathColumns()
- .build();
+ jsonLoaderBuilder = new org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder()
+ .resultSetLoader(loader)
+ .standardOptions(options);
}
@Override
public void eval() {
try {
- jsonReader.setSource(in.start, in.end, in.buffer);
- jsonReader.write(writer);
- buffer = jsonReader.getWorkBuf();
+ jsonLoaderBuilder.fromStream(in.start, in.end, in.buffer);
+ org.apache.drill.exec.store.easy.json.loader.JsonLoader jsonLoader = jsonLoaderBuilder.build();
+ loader.startBatch();
+ jsonLoader.readBatch();
+ loader.close();
Review Comment:
Thanks @vdiravka ! I addressed your change. Once #2568 is merged, I will rebase and hopefully we won't run into any issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [drill] cgivre commented on a diff in pull request #2567: [WIP] DRILL-8239: Convert JSON UDF to EVF
Posted by GitBox <gi...@apache.org>.
cgivre commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r891318837
##########
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConvertFrom.java:
##########
@@ -32,37 +30,45 @@
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
import org.apache.drill.exec.expr.holders.VarBinaryHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;
public class JsonConvertFrom {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JsonConvertFrom.class);
-
private JsonConvertFrom() {
}
@FunctionTemplate(name = "convert_fromJSON", scope = FunctionScope.SIMPLE, isRandom = true)
public static class ConvertFromJson implements DrillSimpleFunc {
@Param VarBinaryHolder in;
- @Inject DrillBuf buffer;
- @Workspace org.apache.drill.exec.vector.complex.fn.JsonReader jsonReader;
+ @Inject
+ ResultSetLoader loader;
+ @Workspace
+ org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder jsonLoaderBuilder;
+
+ @Inject
+ OptionManager options;
@Output ComplexWriter writer;
@Override
public void setup() {
- jsonReader = new org.apache.drill.exec.vector.complex.fn.JsonReader.Builder(buffer)
- .defaultSchemaPathColumns()
- .build();
+ jsonLoaderBuilder = new org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder()
+ .resultSetLoader(loader)
+ .standardOptions(options);
}
@Override
public void eval() {
try {
- jsonReader.setSource(in.start, in.end, in.buffer);
- jsonReader.write(writer);
- buffer = jsonReader.getWorkBuf();
+ jsonLoaderBuilder.fromStream(in.start, in.end, in.buffer);
+ org.apache.drill.exec.store.easy.json.loader.JsonLoader jsonLoader = jsonLoaderBuilder.build();
+ loader.startBatch();
+ jsonLoader.readBatch();
+ loader.close();
Review Comment:
Thanks @vdiravka ! I addressed your change. Once #2568 is merged, are we good to go?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "cgivre (via GitHub)" <gi...@apache.org>.
cgivre commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3814590136
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
You're right, and I was wrong in my earlier answer — thanks for pushing back with the analysis.
The deciding detail is in `ProjectBatchBuilder.addComplexField()`: it calls `initComplexWriters()` unconditionally, for EVF functions as well as legacy `ComplexWriter` ones. So whenever `rsLoaders` is non-empty, `complexWriters` has already been set to a (possibly empty) list, `complexWriters != null` is true, and the `else if` can never be reached. Multi-row queries don't change that — as you say, the branch is decided at setup time, not per batch.
Removed the branch in c5b7424. The loaders don't need a row count there anyway: their row count follows the rows the generated function wrote, and `harvestLoaders()` runs immediately after `setValueCount()`. I also made the `complexWriters` checks use `CollectionUtils.isEmpty()` consistently, since `!= null` vs `isEmpty()` was what made the branch selection confusing in the first place.
To convince myself the loader's default row count limit wasn't quietly relying on that call, I added `testConvertFromJsonBatchLargerThanLoaderRowLimit` — a conversion over `DEFAULT_ROW_COUNT * 2 + 17` rows. It passes with the branch removed.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -28,12 +28,14 @@
import org.apache.drill.exec.physical.config.Project;
import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
import org.apache.drill.exec.record.AbstractSingleRecordBatch;
+import org.apache.drill.exec.record.MaterializedField;
Review Comment:
Fixed in c5b7424 — `JsonConverterUtils` moved up next to the other `org.apache.drill.exec.expr` imports and `MaterializedField` moved after `BatchSchema`.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConverterUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.expr.fn.impl.conv;
+
+
+import io.netty.buffer.DrillBuf;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.store.easy.json.loader.ClosingStreamIterator;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderOptions;
+import org.apache.drill.exec.vector.complex.fn.DrillBufInputStream;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JsonConverterUtils {
+
+ private static final Logger logger = LoggerFactory.getLogger(JsonConverterUtils.class);
+
+ /**
+ * Field name used to wrap the input value so that the record-oriented JSON
+ * loader can read top-level scalars and arrays (not just objects) uniformly.
+ * The single resulting column carries the converted value and is recognised
+ * and unwrapped by {@code ProjectRecordBatch} (which transfers it directly to
+ * the output column instead of wrapping the loader's columns in a map).
+ */
+ public static final String WRAP_FIELD = "drill_json_value_wrapper";
+
+ private static final byte[] WRAP_PREFIX =
+ ("{\"" + WRAP_FIELD + "\":").getBytes(StandardCharsets.UTF_8);
+ private static final byte[] WRAP_SUFFIX = "}".getBytes(StandardCharsets.UTF_8);
+
+ private JsonConverterUtils() {
+ }
+
+ /**
+ * Creates a {@link JsonLoaderImpl} for use in JSON conversion UDFs, using the
+ * system JSON options from the {@link OptionManager}.
+ *
+ * @param rsLoader The {@link ResultSetLoader} used in the UDF
+ * @param options The {@link OptionManager} used in the UDF. This is used to extract the global JSON options
+ * @param stream An input stream containing the input JSON data
+ * @return A {@link JsonLoaderImpl} for use in the UDF.
+ */
+ public static JsonLoaderImpl createJsonLoader(ResultSetLoader rsLoader,
+ OptionManager options,
+ ClosingStreamIterator stream) {
+ JsonLoaderBuilder jsonLoaderBuilder = new JsonLoaderBuilder()
+ .resultSetLoader(rsLoader)
+ .standardOptions(options)
+ .fromStream(() -> stream);
+
+ return (JsonLoaderImpl) jsonLoaderBuilder.build();
+ }
+
+ /**
+ * Creates a {@link JsonLoaderImpl} for use in JSON conversion UDFs, overriding the
+ * {@code allTextMode} and {@code readNumbersAsDouble} options with the values supplied
+ * as function arguments. Remaining options are taken from the system JSON options.
+ *
+ * @param rsLoader The {@link ResultSetLoader} used in the UDF
+ * @param options The {@link OptionManager} used in the UDF. This is used to extract the global JSON options
+ * @param stream An input stream containing the input JSON data
+ * @param allTextMode Whether to read all scalars as text
+ * @param readNumbersAsDouble Whether to read all numbers as doubles
+ * @return A {@link JsonLoaderImpl} for use in the UDF.
+ */
+ public static JsonLoaderImpl createJsonLoader(ResultSetLoader rsLoader,
+ OptionManager options,
+ ClosingStreamIterator stream,
+ boolean allTextMode,
+ boolean readNumbersAsDouble) {
+ JsonLoaderOptions jsonOptions = new JsonLoaderOptions(options);
+ jsonOptions.allTextMode = allTextMode;
+ jsonOptions.readNumbersAsDouble = readNumbersAsDouble;
+
+ JsonLoaderBuilder jsonLoaderBuilder = new JsonLoaderBuilder()
+ .resultSetLoader(rsLoader)
+ .options(jsonOptions)
+ .fromStream(() -> stream);
+
+ return (JsonLoaderImpl) jsonLoaderBuilder.build();
+ }
+
+ /**
+ * Converts a single JSON value (one row of UDF input) into the result set loader.
+ *
+ * <p>Exactly one row is always written to the loader -- even for null/empty
+ * input -- so that the loader's row count stays aligned with the surrounding
+ * project batch. The value is wrapped in a single-field object so that
+ * top-level scalars and arrays parse the same way as objects.</p>
+ *
+ * @param rsLoader the result set loader injected into the UDF
+ * @param jsonLoader the lazily created JSON loader, or {@code null} on the first call
+ * @param options the system JSON options
+ * @param stream the streaming iterator bound to the JSON loader
+ * @param isSet 0 if the input value is null, 1 otherwise
+ * @param start start offset of the value in {@code buffer}
+ * @param end end offset of the value in {@code buffer}
+ * @param buffer buffer holding the input value
+ * @param useArgs whether to honour the explicit allTextMode/readNumbersAsDouble arguments
+ * @param allTextMode explicit all-text-mode argument (used only when {@code useArgs})
+ * @param readNumbersAsDouble explicit read-numbers-as-double argument (used only when {@code useArgs})
+ * @return the JSON loader (created on first use), to be cached by the caller
+ */
+ public static JsonLoaderImpl convertJson(ResultSetLoader rsLoader,
+ JsonLoaderImpl jsonLoader,
+ OptionManager options,
+ ClosingStreamIterator stream,
+ int isSet, int start, int end, DrillBuf buffer,
+ boolean useArgs,
+ boolean allTextMode,
+ boolean readNumbersAsDouble) {
+ RowSetLoader rowWriter = rsLoader.writer();
+ rowWriter.start();
+ // For null or empty input emit an (unset) row to keep the row count aligned.
+ if (isSet == 0 || start == end) {
+ rowWriter.save();
+ return jsonLoader;
+ }
+
+ try {
+ stream.setValue(getWrappedJsonStream(start, end, buffer));
+ if (jsonLoader == null) {
+ jsonLoader = useArgs
+ ? createJsonLoader(rsLoader, options, stream, allTextMode, readNumbersAsDouble)
+ : createJsonLoader(rsLoader, options, stream);
+ }
+ // next() reads the single wrapped record; always save the row so the
+ // count stays aligned even for an empty document.
+ jsonLoader.parser().next();
+ rowWriter.save();
+ } catch (Exception e) {
+ throw UserException.dataReadError(e)
+ .message("Error while reading JSON. ")
+ .addContext(e.getMessage())
+ .build(logger);
+ }
+ return jsonLoader;
+ }
+
+ /**
+ * Wraps the raw input value in a single-field JSON object ({@code {"json": <value>}})
+ * so that the record-oriented JSON loader accepts top-level scalars and arrays.
Review Comment:
Good catch — the Javadoc predated the rename. Updated in c5b7424 to name the real wrapper field and link to `WRAP_FIELD`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [drill] vdiravka commented on a diff in pull request #2567: [WIP] DRILL-8239: Convert JSON UDF to EVF
Posted by GitBox <gi...@apache.org>.
vdiravka commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r891297158
##########
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConvertFrom.java:
##########
@@ -32,37 +30,45 @@
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
import org.apache.drill.exec.expr.holders.VarBinaryHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;
public class JsonConvertFrom {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JsonConvertFrom.class);
-
private JsonConvertFrom() {
}
@FunctionTemplate(name = "convert_fromJSON", scope = FunctionScope.SIMPLE, isRandom = true)
public static class ConvertFromJson implements DrillSimpleFunc {
@Param VarBinaryHolder in;
- @Inject DrillBuf buffer;
- @Workspace org.apache.drill.exec.vector.complex.fn.JsonReader jsonReader;
+ @Inject
+ ResultSetLoader loader;
+ @Workspace
+ org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder jsonLoaderBuilder;
+
+ @Inject
+ OptionManager options;
@Output ComplexWriter writer;
@Override
public void setup() {
- jsonReader = new org.apache.drill.exec.vector.complex.fn.JsonReader.Builder(buffer)
- .defaultSchemaPathColumns()
- .build();
+ jsonLoaderBuilder = new org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder()
+ .resultSetLoader(loader)
+ .standardOptions(options);
}
@Override
public void eval() {
try {
- jsonReader.setSource(in.start, in.end, in.buffer);
- jsonReader.write(writer);
- buffer = jsonReader.getWorkBuf();
+ jsonLoaderBuilder.fromStream(in.start, in.end, in.buffer);
+ org.apache.drill.exec.store.easy.json.loader.JsonLoader jsonLoader = jsonLoaderBuilder.build();
+ loader.startBatch();
+ jsonLoader.readBatch();
+ loader.close();
Review Comment:
It shouldn't be closed in the `eval` part of the UDF, because `loader` is using in `ProjectRecordBatch` and creating the function output. And it is actually is closing in `ProjectRecordBatch` after #2568 merged
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "shfshihuafeng (via GitHub)" <gi...@apache.org>.
shfshihuafeng commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3781868274
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
In my opinion, Multi-row queries don't change the branch logic here. Whether complexWriters is null is decided at setup time (by addComplexField() → initComplexWriters()), not per-batch.
Are you referring to a multi‑row scenario like following example? Query reads 10,000 rows from test_multirow.json
and, for each row, concatenates id and name into a JSON string, then converts it with convert_fromJSON. Does that match what you meant by "multi‑row queries"?
```
{"id":6,"name":"user_6","value":60}
apache drill> SELECT
2..semicolon> convert_fromJSON(CONCAT('{"nested_id":', CAST(id AS VARCHAR), ',"orig_name":"', name, '"}')) AS js
3..semicolon> FROM
4..semicolon> dfs.test.`test_multirow.json`;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "cgivre (via GitHub)" <gi...@apache.org>.
cgivre merged PR #2567:
URL: https://github.com/apache/drill/pull/2567
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "shfshihuafeng (via GitHub)" <gi...@apache.org>.
shfshihuafeng commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3770696057
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -28,12 +28,14 @@
import org.apache.drill.exec.physical.config.Project;
import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
import org.apache.drill.exec.record.AbstractSingleRecordBatch;
+import org.apache.drill.exec.record.MaterializedField;
Review Comment:
nit: The imports for MaterializedField and JsonConverterUtils break the alphabetical ordering of the other imports.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
In my opinion, the loop body calling setTargetRowCount in ProjectRecordBatch.setValueCount() is dead code.
Two cases exist:
With a ComplexWriter function (convert_fromJSON, split, etc.): addComplexField() is called → complexWriters is non-null → complexWriters != null is true → enters the first branch, else if is short‑circuited.
Without any ComplexWriter function (e.g. select *): complexWriters stays null → complexWriters != null is false → the else if condition is evaluated. But addLoader() is also not called → rsLoaders is null → !CollectionUtils.isEmpty(rsLoaders) is false → the loop body is not entered.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/JsonConverterUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.expr.fn.impl.conv;
+
+
+import io.netty.buffer.DrillBuf;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.store.easy.json.loader.ClosingStreamIterator;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderImpl.JsonLoaderBuilder;
+import org.apache.drill.exec.store.easy.json.loader.JsonLoaderOptions;
+import org.apache.drill.exec.vector.complex.fn.DrillBufInputStream;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JsonConverterUtils {
+
+ private static final Logger logger = LoggerFactory.getLogger(JsonConverterUtils.class);
+
+ /**
+ * Field name used to wrap the input value so that the record-oriented JSON
+ * loader can read top-level scalars and arrays (not just objects) uniformly.
+ * The single resulting column carries the converted value and is recognised
+ * and unwrapped by {@code ProjectRecordBatch} (which transfers it directly to
+ * the output column instead of wrapping the loader's columns in a map).
+ */
+ public static final String WRAP_FIELD = "drill_json_value_wrapper";
+
+ private static final byte[] WRAP_PREFIX =
+ ("{\"" + WRAP_FIELD + "\":").getBytes(StandardCharsets.UTF_8);
+ private static final byte[] WRAP_SUFFIX = "}".getBytes(StandardCharsets.UTF_8);
+
+ private JsonConverterUtils() {
+ }
+
+ /**
+ * Creates a {@link JsonLoaderImpl} for use in JSON conversion UDFs, using the
+ * system JSON options from the {@link OptionManager}.
+ *
+ * @param rsLoader The {@link ResultSetLoader} used in the UDF
+ * @param options The {@link OptionManager} used in the UDF. This is used to extract the global JSON options
+ * @param stream An input stream containing the input JSON data
+ * @return A {@link JsonLoaderImpl} for use in the UDF.
+ */
+ public static JsonLoaderImpl createJsonLoader(ResultSetLoader rsLoader,
+ OptionManager options,
+ ClosingStreamIterator stream) {
+ JsonLoaderBuilder jsonLoaderBuilder = new JsonLoaderBuilder()
+ .resultSetLoader(rsLoader)
+ .standardOptions(options)
+ .fromStream(() -> stream);
+
+ return (JsonLoaderImpl) jsonLoaderBuilder.build();
+ }
+
+ /**
+ * Creates a {@link JsonLoaderImpl} for use in JSON conversion UDFs, overriding the
+ * {@code allTextMode} and {@code readNumbersAsDouble} options with the values supplied
+ * as function arguments. Remaining options are taken from the system JSON options.
+ *
+ * @param rsLoader The {@link ResultSetLoader} used in the UDF
+ * @param options The {@link OptionManager} used in the UDF. This is used to extract the global JSON options
+ * @param stream An input stream containing the input JSON data
+ * @param allTextMode Whether to read all scalars as text
+ * @param readNumbersAsDouble Whether to read all numbers as doubles
+ * @return A {@link JsonLoaderImpl} for use in the UDF.
+ */
+ public static JsonLoaderImpl createJsonLoader(ResultSetLoader rsLoader,
+ OptionManager options,
+ ClosingStreamIterator stream,
+ boolean allTextMode,
+ boolean readNumbersAsDouble) {
+ JsonLoaderOptions jsonOptions = new JsonLoaderOptions(options);
+ jsonOptions.allTextMode = allTextMode;
+ jsonOptions.readNumbersAsDouble = readNumbersAsDouble;
+
+ JsonLoaderBuilder jsonLoaderBuilder = new JsonLoaderBuilder()
+ .resultSetLoader(rsLoader)
+ .options(jsonOptions)
+ .fromStream(() -> stream);
+
+ return (JsonLoaderImpl) jsonLoaderBuilder.build();
+ }
+
+ /**
+ * Converts a single JSON value (one row of UDF input) into the result set loader.
+ *
+ * <p>Exactly one row is always written to the loader -- even for null/empty
+ * input -- so that the loader's row count stays aligned with the surrounding
+ * project batch. The value is wrapped in a single-field object so that
+ * top-level scalars and arrays parse the same way as objects.</p>
+ *
+ * @param rsLoader the result set loader injected into the UDF
+ * @param jsonLoader the lazily created JSON loader, or {@code null} on the first call
+ * @param options the system JSON options
+ * @param stream the streaming iterator bound to the JSON loader
+ * @param isSet 0 if the input value is null, 1 otherwise
+ * @param start start offset of the value in {@code buffer}
+ * @param end end offset of the value in {@code buffer}
+ * @param buffer buffer holding the input value
+ * @param useArgs whether to honour the explicit allTextMode/readNumbersAsDouble arguments
+ * @param allTextMode explicit all-text-mode argument (used only when {@code useArgs})
+ * @param readNumbersAsDouble explicit read-numbers-as-double argument (used only when {@code useArgs})
+ * @return the JSON loader (created on first use), to be cached by the caller
+ */
+ public static JsonLoaderImpl convertJson(ResultSetLoader rsLoader,
+ JsonLoaderImpl jsonLoader,
+ OptionManager options,
+ ClosingStreamIterator stream,
+ int isSet, int start, int end, DrillBuf buffer,
+ boolean useArgs,
+ boolean allTextMode,
+ boolean readNumbersAsDouble) {
+ RowSetLoader rowWriter = rsLoader.writer();
+ rowWriter.start();
+ // For null or empty input emit an (unset) row to keep the row count aligned.
+ if (isSet == 0 || start == end) {
+ rowWriter.save();
+ return jsonLoader;
+ }
+
+ try {
+ stream.setValue(getWrappedJsonStream(start, end, buffer));
+ if (jsonLoader == null) {
+ jsonLoader = useArgs
+ ? createJsonLoader(rsLoader, options, stream, allTextMode, readNumbersAsDouble)
+ : createJsonLoader(rsLoader, options, stream);
+ }
+ // next() reads the single wrapped record; always save the row so the
+ // count stays aligned even for an empty document.
+ jsonLoader.parser().next();
+ rowWriter.save();
+ } catch (Exception e) {
+ throw UserException.dataReadError(e)
+ .message("Error while reading JSON. ")
+ .addContext(e.getMessage())
+ .build(logger);
+ }
+ return jsonLoader;
+ }
+
+ /**
+ * Wraps the raw input value in a single-field JSON object ({@code {"json": <value>}})
+ * so that the record-oriented JSON loader accepts top-level scalars and arrays.
Review Comment:
nit: getWrappedJsonStream Javadoc says {"json": <value>} but the actual field is drill_json_value_wrapper (WRAP_FIELD).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "shfshihuafeng (via GitHub)" <gi...@apache.org>.
shfshihuafeng commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3771229111
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
**What use case would actually cause this loop body to execute?**
In my opinion, the loop body calling setTargetRowCount in ProjectRecordBatch.setValueCount() is dead code.
Two cases exist:
With a ComplexWriter function (convert_fromJSON, split, etc.): addComplexField() is called → complexWriters is non-null( initComplexWriters()->projectBatch.complexWriters = new ArrayList<>();) → complexWriters != null is true → enters the first branch, else if is short‑circuited.
Without any ComplexWriter function (e.g. select *): complexWriters stays null → complexWriters != null is false → the else if condition is evaluated. But addLoader() is also not called → rsLoaders is null → !CollectionUtils.isEmpty(rsLoaders) is false → the loop body is not entered.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] DRILL-8239: Convert JSON UDF to EVF (drill)
Posted by "cgivre (via GitHub)" <gi...@apache.org>.
cgivre commented on code in PR #2567:
URL: https://github.com/apache/drill/pull/2567#discussion_r3776513971
##########
exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/project/ProjectRecordBatch.java:
##########
@@ -274,8 +338,10 @@ private void setValueCount(int count) {
for (ComplexWriter writer : complexWriters) {
writer.setValueCount(count);
}
- } else if (rsLoader != null) {
- rsLoader.setTargetRowCount(count);
+ } else if (!CollectionUtils.isEmpty(rsLoaders)) {
+ for (ResultSetLoader loader : rsLoaders) {
+ loader.setTargetRowCount(count);
Review Comment:
I didn't originally write this, so I'm not 100% sure, but if I remember correctly, this was necessary for multi-row queries.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@drill.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org