You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@hop.apache.org by "bamaer (via GitHub)" <gi...@apache.org> on 2026/09/30 17:24:03 UTC
[PR] Fixes #2929 : Stream Enhanced JSON Output to the file (hop)
bamaer opened a new pull request, #8713:
URL: https://github.com/apache/hop/pull/8713
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] Run `mvn clean install apache-rat:check` to make sure basic checks pass. A more thorough check will be performed on your pull request automatically.
- [x] If you have a group of commits related to the same change, please squash your commits into one and force push your branch using `git rebase -i`.
- [x] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable.
To make clear that you license your contribution under the [Apache License Version 2.0, January 2004](http://www.apache.org/licenses/LICENSE-2.0)
you have to acknowledge this by using the following check-box.
- [x] I hereby declare this contribution to be licensed under the [Apache License Version 2.0, January 2004](http://www.apache.org/licenses/LICENSE-2.0)
- [ ] In any other case, please file an [Apache Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
--
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #2929 : Stream Enhanced JSON Output to the file (hop)
Posted by "bamaer (via GitHub)" <gi...@apache.org>.
bamaer merged PR #8713:
URL: https://github.com/apache/hop/pull/8713
--
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fixes #2929 : Stream Enhanced JSON Output to the file (hop)
Posted by "mattcasters (via GitHub)" <gi...@apache.org>.
mattcasters commented on code in PR #8713:
URL: https://github.com/apache/hop/pull/8713#discussion_r4153888738
##########
plugins/transforms/json/src/main/java/org/apache/hop/pipeline/transforms/jsonoutputenhanced/JsonEOutput.java:
##########
@@ -321,113 +306,187 @@ 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 = new Object[2];
- data.jsonLength = data.jsonSerialized.length();
+ additionalRowFields[0] = data.jsonSerialized;
- if (data.outputRowMeta != null) {
+ // Fill accessory fields
+ if (!Utils.isEmpty(meta.getJsonSizeFieldName())) {
+ additionalRowFields[1] = data.jsonLength;
+ }
- Object[] keyRow = new Object[meta.getKeyFields().size()];
+ Object[] outputRowData = RowDataUtil.addRowData(keyRow, keyRow.length, additionalRowFields);
+ incrementLinesOutput();
- // Create a new object with specified fields
- if (data.isWriteToFile) {
- globalItemNode = new ObjectNode(nc);
- }
+ putRow(data.outputRowMeta, outputRowData);
- 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);
- }
- }
+ // Data are safe
+ data.rowsAreSafe = true;
+ }
- 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);
- }
- data.jsonItems.add(globalItemNode);
+ 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;
+ }
- Object[] additionalRowFields = new Object[2];
+ /**
+ * 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;
+ }
- additionalRowFields[0] = data.jsonSerialized;
+ /** 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);
+ }
- // Fill accessory fields
- if (!Utils.isEmpty(meta.getJsonSizeFieldName())) {
- additionalRowFields[1] = data.jsonLength;
+ /**
+ * 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 {
+ try {
+ if (data.fileItemCount == 0) {
+ data.pendingFileItem = item;
+ } else {
+ if (data.fileItemCount == 1) {
+ startFileDocument(true);
+ data.fileGenerator.writeTree(data.pendingFileItem);
+ data.pendingFileItem = null;
+ }
+ data.fileGenerator.writeTree(item);
}
-
- Object[] outputRowData = RowDataUtil.addRowData(keyRow, keyRow.length, additionalRowFields);
+ } catch (IOException e) {
+ throw new HopTransformException(BaseMessages.getString(PKG, "JsonOutput.Error.Writing"), e);
+ }
+ data.fileItemCount++;
+ if (!data.isOutputValue) {
incrementLinesOutput();
-
- putRow(data.outputRowMeta, outputRowData);
}
- // Data are safe
- data.rowsAreSafe = true;
+ int splitOutputAfter = meta.getFileSettings().getSplitOutputAfter();
+ if (splitOutputAfter > 0 && data.fileItemCount >= splitOutputAfter) {
+ finishFile();
+ }
}
- private void writeJsonFile() throws HopTransformException {
- // Open a file
- if (data.isWriteToFile && !openNewFile())
+ private void startFileDocument(boolean array) throws IOException, HopTransformException {
+ if (!openNewFile()) {
throw new HopTransformException(
BaseMessages.getString(PKG, "JsonOutput.Error.OpenNewFile", buildFilename()));
- // Write data to file
+ }
+ data.fileGenerator = fileMapper.getFactory().createGenerator(data.writer);
+ // The file is closed separately, with its lineage
+ data.fileGenerator.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET);
Review Comment:
**[bug]** The generator is created with `FLUSH_AFTER_WRITE_VALUE` disabled and `AUTO_CLOSE_TARGET` turned off, and it is closed only in `finishFile()`. `dispose()` still only calls `closeFile()`. `finishFile()` runs when `getRow()` returns null, but not when `processRow()` throws (bad fragment, disk full, a later conversion error). On that path the generator's buffer is never flushed, `closeFile()` closes the writer underneath it, and the file is missing its tail and its closing array or object. A short write-to-file run can lose every item, because those `writeTree` calls never filled the generator buffer.
**Suggestion:** In `dispose()`, if `data.fileGenerator != null`, close it before `closeFile()`, then set it and `pendingFileItem` to null. `close()` flushes the buffer, and the default `AUTO_CLOSE_JSON_CONTENT` finishes an open array or object so the partial file is at least well-formed. Leave `finishFile()` as the success path that also resets `fileItemCount`.
--
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: issues-unsubscribe@hop.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org