Writing a Delta Lake checkpoint fails for a table with a multi-field struct column

Trino version

484-SNAPSHOT (master, 9992d51)

Please describe the bug

Writing a Delta Lake checkpoint fails with IllegalStateException when the transaction log holds JSON statistics for a struct column with more than one field.

DeltaLakeParquetStatisticsUtils.jsonValueToTrinoValue iterates the struct fields like this:

Map<?, ?> values = (Map<?, ?>) jsonValue;
return buildRowValue(rowType, fields -> {
    for (int i = 0; i < values.size(); ++i) {
        ...
        Object fieldValue = jsonValueToTrinoValue(fieldType, values.remove(fieldName));
        ...
    }
    checkState(values.isEmpty(), "All fields must be converted into Trino value: %s", values);
});

values.size() is re-evaluated on every iteration while values.remove(fieldName) shrinks the map, so the loop stops after roughly half the fields. It should iterate over the field count of the row type instead. values is also the map Jackson produced, so it is mutated in place.

Statistics are written as a struct by default (delta.checkpoint.writeStatsAsStruct), so any table with a multi-field struct column reaches this. A single-field struct gets through the loop, and TestCheckpointWriter#testCheckpointWriteReadJsonRoundtrip sets delta.checkpoint.writeStatsAsStruct=false, so its statistics never go through this path - which is why the existing tests do not catch it.

Reproduction. Writing a checkpoint for an add entry whose schema has row struct<s1: integer, s2: string> and whose statistics are {"numRecords":20,"minValues":{"row":{"s1":1,"s2":"a"}},"maxValues":{"row":{"s1":2,"s2":"z"}},"nullCount":{"row":{"s1":0,"s2":1}}}:

java.lang.IllegalStateException: All fields must be converted into Trino value: {s2=a}
	at io.trino.plugin.deltalake.transactionlog.DeltaLakeParquetStatisticsUtils.lambda$jsonValueToTrinoValue$0(DeltaLakeParquetStatisticsUtils.java:165)
	at io.trino.spi.block.BufferedRowValueBuilder.build(BufferedRowValueBuilder.java:66)
	at io.trino.spi.block.RowValueBuilder.buildRowValue(RowValueBuilder.java:26)
	at io.trino.plugin.deltalake.transactionlog.DeltaLakeParquetStatisticsUtils.jsonValueToTrinoValue(DeltaLakeParquetStatisticsUtils.java:158)
	at io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriter.preprocessMinMaxValues(CheckpointWriter.java:492)
	at io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriter.writeMinMaxMapAsFields(CheckpointWriter.java:463)
	at io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriter.writeParsedStats(CheckpointWriter.java:416)
	at io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriter.writeAddFileEntry(CheckpointWriter.java:302)
	at io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriter.write(CheckpointWriter.java:172)

(intermediate lambda$ and JDK frames removed)

Fixing the loop is necessary but not sufficient. #24030 changes this loop to iterate over the field count, but converting struct statistics from JSON then fails further along: @Pluies reported it on that pull request (comment) with ClassCastException: class java.util.LinkedHashMap cannot be cast to class io.trino.spi.block.SqlRow in writeNullCountAsFields, and @pajaks replied that mapping a row type from JSON to Parquet is not supported and is something outside that pull request; the test there works around it by dropping the row column from the JSON statistics. So this issue should not be closed by #24030 alone, and the later failure is not about the field count.