mattcasters commented on code in PR #8715:
URL: https://github.com/apache/hop/pull/8715#discussion_r4153876935
##########
plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java:
##########
@@ -321,116 +317,254 @@ private String
getKeyJsonAttributeName(JsonEOutputKeyField field) {
return Const.NVL(elementName, field.getFieldName());
}
+ /**
+ * A group is complete: send its row to the output field and, when group
keys are used, write its
+ * item to the file.
+ */
+ private void finishGroup(Object[] groupRow) throws HopException {
+ if (Utils.isEmpty(data.jsonKeyGroupItems)) {
+ return;
+ }
+ if (isDebug()) {
+ logDebug("Record Num: " + data.nrRow + " - Generating JSON chunk");
+ }
+ if (data.isOutputValue) {
+ outputRow(groupRow);
+ }
+ if (data.isWriteToFile && !data.streamFileRows) {
+ if (meta.getKeyFields().isEmpty()) {
+ // All rows are merged into a single item
+ for (ObjectNode item : data.jsonKeyGroupItems) {
+ writeFileItem(item);
+ }
+ } else {
+ writeFileItem(buildGroupFileItem(groupRow));
+ }
+ }
+ data.jsonKeyGroupItems = new ArrayList<>();
+ }
+
private void outputRow(Object[] rowData) throws HopException {
- // We can now output an object
- ObjectNode globalItemNode = null;
+ serializeJson(data.jsonKeyGroupItems);
+ data.jsonLength = data.jsonSerialized.length();
- if (Utils.isEmpty(data.jsonKeyGroupItems)) return;
+ Object[] keyRow = getKeyValues(rowData);
- if (!data.jsonKeyGroupItems.isEmpty()) {
- serializeJson(data.jsonKeyGroupItems);
- }
+ Object[] additionalRowFields =
+ Utils.isEmpty(meta.getJsonSizeFieldName())
+ ? new Object[] {data.jsonSerialized}
+ : new Object[] {data.jsonSerialized, data.jsonLength};
- data.jsonLength = data.jsonSerialized.length();
+ Object[] outputRowData = RowDataUtil.addRowData(keyRow, keyRow.length,
additionalRowFields);
+ // addRowData over-allocates. The published row matches the metadata width.
+ if (outputRowData.length != data.outputRowMeta.size()) {
+ outputRowData = Arrays.copyOf(outputRowData, data.outputRowMeta.size());
+ }
+ incrementLinesOutput();
- if (data.outputRowMeta != null) {
+ putRow(data.outputRowMeta, outputRowData);
- Object[] keyRow = new Object[meta.getKeyFields().size()];
+ // Data are safe
+ data.rowsAreSafe = true;
+ }
- // Create a new object with specified fields
- if (data.isWriteToFile) {
- globalItemNode = new ObjectNode(nc);
+ private Object[] getKeyValues(Object[] rowData) throws HopException {
+ Object[] keyRow = new Object[meta.getKeyFields().size()];
+ for (int i = 0; i < meta.getKeyFields().size(); i++) {
+ JsonEOutputKeyField keyField = meta.getKeyFields().get(i);
+ try {
+ IValueMeta vmi =
data.inputRowMeta.getValueMeta(data.keysGroupIndexes[i]);
+ keyRow[i] =
+ switch (vmi.getType()) {
+ case IValueMeta.TYPE_BOOLEAN ->
+ data.inputRowMeta.getBoolean(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_INTEGER ->
+ data.inputRowMeta.getInteger(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_NUMBER ->
+ data.inputRowMeta.getNumber(rowData,
data.keysGroupIndexes[i]);
+ case IValueMeta.TYPE_BIGNUMBER ->
+ data.inputRowMeta.getBigNumber(rowData,
data.keysGroupIndexes[i]);
+ default -> data.inputRowMeta.getString(rowData,
data.keysGroupIndexes[i]);
+ };
+ } catch (HopValueException e) {
+ throw new HopException(
+ "Error getting json values for key field: " +
keyField.getFieldName(), e);
}
+ }
+ return keyRow;
+ }
- for (int i = 0; i < meta.getKeyFields().size(); i++) {
- JsonEOutputKeyField keyField = meta.getKeyFields().get(i);
- try {
- IValueMeta vmi =
data.inputRowMeta.getValueMeta(data.keysGroupIndexes[i]);
- switch (vmi.getType()) {
- case IValueMeta.TYPE_BOOLEAN:
- keyRow[i] = data.inputRowMeta.getBoolean(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField),
(Boolean) keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_INTEGER:
- keyRow[i] = data.inputRowMeta.getInteger(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (Long)
keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_NUMBER:
- keyRow[i] = data.inputRowMeta.getNumber(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (Double)
keyRow[i]);
- }
- break;
- case IValueMeta.TYPE_BIGNUMBER:
- keyRow[i] = data.inputRowMeta.getBigNumber(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField),
(BigDecimal) keyRow[i]);
- }
- break;
- default:
- keyRow[i] = data.inputRowMeta.getString(rowData,
data.keysGroupIndexes[i]);
- if (data.isWriteToFile) {
- globalItemNode.put(getKeyJsonAttributeName(keyField), (String)
keyRow[i]);
- }
- break;
- }
- } catch (HopValueException e) {
- throw new HopException(
- "Error getting json values for key field: " +
keyField.getFieldName(), e);
- }
+ /**
+ * The file item of a group: the key fields plus the group's items under the
output value name.
+ * The JSON block name only wraps the file, not every group.
+ */
+ private ObjectNode buildGroupFileItem(Object[] groupRow) throws HopException
{
+ ObjectNode groupItem = new ObjectNode(nc);
+ Object[] keyRow = getKeyValues(groupRow);
+ for (int i = 0; i < keyRow.length; i++) {
+ String name = getKeyJsonAttributeName(meta.getKeyFields().get(i));
+ switch (keyRow[i]) {
+ case null -> groupItem.putNull(name);
+ case Boolean b -> groupItem.put(name, b);
+ case Long l -> groupItem.put(name, l);
+ case Double d -> groupItem.put(name, d);
+ case BigDecimal bd -> groupItem.put(name, bd);
+ default -> groupItem.put(name, keyRow[i].toString());
}
+ }
+ groupItem.set(meta.getOutputValue(),
buildGroupValue(data.jsonKeyGroupItems));
+ return groupItem;
+ }
- if (data.isWriteToFile) {
- try {
- // JSON serialization here...
- JsonNode jsonNode = mapper.readTree(data.jsonSerialized);
- if (meta.getOutputValue() != null) {
- globalItemNode.set(meta.getOutputValue(), jsonNode);
- }
- } catch (IOException e) {
- throw new HopException("Error serializing JSON values", e);
+ /** The items of a group: an array, or the single item unless arrays are
forced. */
+ private JsonNode buildGroupValue(List<ObjectNode> items) {
+ if (items.size() > 1 || meta.isUseArrayWithSingleInstance()) {
+ return new ArrayNode(nc).addAll(items);
+ }
+ return items.get(0);
+ }
+
+ /**
+ * Write an item to the file straight away, so the file never has to fit in
memory. The first item
+ * is held back: a file with a single item holds that item, not an array,
unless arrays are
+ * forced.
+ */
+ private void writeFileItem(JsonNode item) throws HopException {
+ if (meta.isNewlineDelimited()) {
+ writeNdJsonItem(item);
+ return;
+ }
+ try {
+ if (data.fileItemCount == 0) {
+ data.pendingFileItem = item;
+ } else {
+ if (data.fileItemCount == 1) {
+ startFileDocument(true);
+ data.fileGenerator.writeTree(data.pendingFileItem);
+ data.pendingFileItem = null;
}
- data.jsonItems.add(globalItemNode);
+ data.fileGenerator.writeTree(item);
}
+ } catch (IOException e) {
+ throw new HopTransformException(BaseMessages.getString(PKG,
"JsonOutput.Error.Writing"), e);
+ }
+ data.fileItemCount++;
+ if (!data.isOutputValue) {
+ incrementLinesOutput();
+ }
- Object[] additionalRowFields = new Object[2];
+ int splitOutputAfter = meta.getFileSettings().getSplitOutputAfter();
+ if (splitOutputAfter > 0 && data.fileItemCount >= splitOutputAfter) {
+ finishFile();
+ }
+ }
- additionalRowFields[0] = data.jsonSerialized;
+ private void startFileDocument(boolean array) throws IOException,
HopTransformException {
+ if (!openNewFile()) {
+ throw new HopTransformException(
+ BaseMessages.getString(PKG, "JsonOutput.Error.OpenNewFile",
buildFilename()));
+ }
+ data.fileGenerator = fileMapper.getFactory().createGenerator(data.writer);
+ // The file is closed separately, with its lineage
+ data.fileGenerator.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET);
+ if (meta.isJsonPrettified()) {
+ data.fileGenerator.setPrettyPrinter(new DefaultPrettyPrinter());
+ }
+ if (!Utils.isEmpty(meta.getJsonBloc())) {
+ data.fileGenerator.writeStartObject();
+ data.fileGenerator.writeFieldName(meta.getJsonBloc());
Review Comment:
**[suggestion]** The block name is not resolved here, and `finishFile()`
uses the same raw `meta.getJsonBloc()` when it closes the object.
`writeNdJsonItem()` resolves it (`resolve(meta.getJsonBloc())`), and `init()`
already stores that resolved value in `data.realBlocName` without reading it.
`serializeJson()` also writes the raw name into the output field.
A block name of `${BLOC}` is therefore a literal key in a normal file and in
the output field, and the resolved key on each NDJSON line. With operation type
"both", the field and the file do not match.
**Suggestion:** Use `data.realBlocName` for `startFileDocument`,
`finishFile`, `writeNdJsonItem`, and `serializeJson`. Literal names stay
unchanged, and both file formats agree with the output field.
##########
plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java:
##########
@@ -613,6 +745,7 @@ public boolean openNewFile() {
String filename = buildFilename();
createParentFolder(filename);
+ validateNdJsonAppend(filename);
Review Comment:
**[suggestion]** `validateNdJsonAppend` throws
`JsonEOutput.Error.NdJsonAppendBoundary` before the output stream is opened,
which correctly avoids truncating the existing file. `openNewFile` then catches
every `Exception`, logs it, and returns false. `init()` and `writeNdJsonItem()`
turn that false into the generic "Could not open file" error, so the boundary
reason is only in the log and looks the same as a permissions or path failure.
**Suggestion:** Let that `IOException` leave `openNewFile`, or catch it and
throw `HopTransformException` with the same message, before the generic open
failure is reported.
##########
plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java:
##########
@@ -530,28 +663,27 @@ private void initDataFieldsPositionsArray() throws
HopException {
}
}
- private boolean initKeyFieldsPositionArray(Object[] r) {
- data.keysGroupIndexes = new int[meta.getKeyFields().size()];
-
- for (int i = 0; i < meta.getKeyFields().size(); i++) {
- data.keysGroupIndexes[i] =
-
data.inputRowMeta.indexOfValue(meta.getKeyFields().get(i).getFieldName());
- if ((r != null) && (data.keysGroupIndexes[i] < 0)) {
- setErrors(1);
- stopAll();
- return true;
- }
- }
- return false;
- }
-
@Override
public void dispose() {
if (data.jsonKeyGroupItems != null) {
data.jsonKeyGroupItems = null;
}
+ // A cancelled or failed row can leave the generator open. Close it before
the writer.
+ // Do not finish the group: a partial file is not an atomic write.
+ if (data.fileGenerator != null) {
Review Comment:
**[bug]** A non-NDJSON file parks the first completed item in
`pendingFileItem` and only creates `fileGenerator` when the second item arrives
(`writeFileItem`). This branch closes an open generator and then `closeFile()`
runs, but the parked item is never written. `fileItemCount` and
`incrementLinesOutput()` were already updated when the item was parked, so a
stop or a failure on the next row reports a written line while the file stays
empty. If the file was opened during `init()`, that open already truncated it.
This is not the partial group the comment is avoiding. The parked value is a
finished row or a finished group, and the group still being built stays in
`jsonKeyGroupItems` until `finishGroup`. `finishFile()` already writes a single
pending item the same way a normal end-of-input does, and it returns
immediately when `fileItemCount` is 0.
**Suggestion:** From `dispose()`, if `fileItemCount > 0`, call
`finishFile()` and log a failure instead of only closing the generator. Then
call `closeFile()` so a file opened at init with no items is still closed. A
normal end-of-input has already set the count to 0, so this does not write the
document twice.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
