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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
