The sorry state of Java deserialization

create table measurements(
   city varchar, 
   temperature int
)

try (var citiesDis = new DataInputStream(
                         new BufferedInputStream(Files.newInputStream(citiesFile)));
     var temperaturesDis = new DataInputStream(
                         new BufferedInputStream(Files.newInputStream(temperaturesFile)))) 
{

    while (true) {
        int temperature = temperaturesDis.readShort();
        String city = citiesDis.readUTF();
        observers.computeIfAbsent(city, 
                         k -> new ResultObserver())
                      .observe(temperature / 100.);
    }
}
catch (EOFException ex) {} // ignore

try (var queryStmt = conn.createStatement();
     var rs = queryStmt.executeQuery(
         "select city, temperature from measurements"))
{
    queryStmt.setFetchSize(1000);
    while (rs.next()) {
        observerMap.computeIfAbsent(rs.getString(1), 
                       k -> new ResultsObserver())
                   .observe(rs.getInt(2) / 100.);
    }
}

message OneBrcProtoRecord {
  string city = 1;
  int32 temperature = 2;
}

try (var is = new BufferedInputStream(
                    Files.newInputStream(tempFile))) {
    for (int i = 0; i < records; i++) {
        Onebrc.OneBrcProtoRecord record = 
               Onebrc.OneBrcProtoRecord.parseDelimitedFrom(is);
        stats.computeIfAbsent(record.getCity(), 
                  k -> new ResultsObserver())
             .observe(record.getTemperature());
    }
}

ByteBuffer buffer = ByteBuffer.allocate(4096); // 4K block size to match the disk

try (var fc = (FileChannel) Files.newByteChannel(tempFile, StandardOpenOption.READ)) {

    for (int i = 0; i < records; i++) {

        int size;
        if (!buffer.hasRemaining()) {
            buffer.clear();
            fc.read(buffer);
            buffer.flip();
        }
        size = buffer.get();

        if (buffer.remaining() < size) {
            buffer.compact();
            fc.read(buffer);
            buffer.flip();
        }

        var cis = CodedInputStream.newInstance(
                     buffer.slice(buffer.position(), size));
        var record = Onebrc.OneBrcProtoRecord.parseFrom(cis);
        buffer.position(buffer.position() + size);

        stats.computeIfAbsent(record.getCity(), k -> new ResultsObserver())
             .observe(record.getTemperature());
    }

try (var stream = ParquetReader.streamContent(tempFile.toFile(),
        HydratorSupplier.constantly(new OneBRCParquetHydrator()))) 
{
    stream.collect(Collectors.groupingBy(r -> r.city, 
                   Collectors.summarizingDouble(r -> r.temperature / 100.)))
            .forEach(...);
}

public static class OneBRCParquetRecord {
    public String city;
    public int temperature;

    public OneBRCParquetRecord() {}
    public OneBRCParquetRecord(String city, int temperature) {
        this.city = city;
        this.temperature = temperature;
    }

    public static MessageType schema = new MessageType(
            OneBRCParquetRecord.class.getSimpleName(),
            Types.required(BINARY).as(stringType()).named("city"),
            Types.required(INT32).named("temperature")
    );


    public void dehydrate(ValueWriter valueWriter) {
        valueWriter.write("city", city);
        valueWriter.write("temperature", temperature);
    }

    public OneBRCParquetRecord add(String heading, Object value) {
        if ("city".equals(heading)) {
            city = (String) value;
        } else if ("temperature".equals(heading)) {
            temperature = (Integer) value;
        } else {
            throw new UnsupportedOperationException("Unknown heading '" + heading + "'");
        }

        return this;
    }
}

static class OneBRCParquetHydrator implements Hydrator<OneBRCParquetRecord, OneBRCParquetRecord> {

    @Override
    public OneBRCParquetRecord start() {
        return new OneBRCParquetRecord();
    }

    @Override
    public OneBRCParquetRecord add(OneBRCParquetRecord target, String heading, Object value) {
        return target.add(heading, value);
    }

    @Override
    public OneBRCParquetRecord finish(OneBRCParquetRecord target) {
        return target;
    }

}

try (var ois = new ObjectInputStream(new BufferedInputStream(Files.newInputStream(data)))) {
    for (;;) {
        if (!(ois.readObject() instanceof JavaSerializableMeasurement jsm))
           continue;
		observerMap.computeIfAbsent(jsm.city, 
						k -> new ResultsObserver())
				   .observe(jsm.temperature / 100.);
	
    }
}
catch (EOFException e) {}

// ...

record JavaSerializableMeasurement(
  String city, 
  int temperature
) implements Serializable {}

var buffer = ByteBuffer.allocate(4096);
try (var fc = 
	 (FileChannel) Files.newByteChannel(tempFile, 
                        StandardOpenOption.READ)) 
{

    buffer.flip();

    for (int i = 0; i < records; i++) {

        if (buffer.remaining() < 32) {
            buffer.compact();
            fc.read(buffer);
            buffer.flip();
        }

        int len = buffer.get();
        byte[] cityBytes = new byte[len];
        buffer.get(cityBytes);
        String city = new String(cityBytes);
        int temperature = buffer.getShort();

        stats.computeIfAbsent(city, k -> new ResultsObserver())
             .observe(temperature / 100.);
    }

List<String> cities = Files.readAllLines(citiesPath);

// using an array over a map saves about 6-7 seconds runtime
// via better locality
ResultsObserver[] stats = new ResultsObserver[cities.size()];
Arrays.setAll(stats, i -> new ResultsObserver());

try (var fc = (FileChannel) Files.newByteChannel(tempFile, StandardOpenOption.READ)) {

    for (int i = 0; i < records; i++) {

        if (buffer.remaining() < 3) {
            buffer.compact();
            fc.read(buffer);
            buffer.flip();
        }

        int cityOrd = buffer.get();
        int temperature = buffer.getShort();
        stats[cityOrd].observe(temperature / 100.);

// ...

try (var arena = Arena.ofConfined();
     var fcCities = (FileChannel) Files.newByteChannel(tempFile1, StandardOpenOption.READ);
     var fcMeasurements = (FileChannel) Files.newByteChannel(tempFile2, StandardOpenOption.READ)
) {
    var citiesSegment = 
              fcCities.map(FileChannel.MapMode.READ_ONLY, 
                           0, fcCities.size(), arena);
    var measurementsSegment = 
              fcMeasurements.map(FileChannel.MapMode.READ_ONLY, 
                                 0, fcMeasurements.size(), arena);


    for (int i = 0; i < records; i++) {
        int cityOrd 
            = citiesSegment.getAtIndex(ValueLayout.JAVA_BYTE, i);
        int temperature 
            = measurementsSegment.getAtIndex(ValueLayout.JAVA_SHORT, i);
            
        stats[cityOrd].observe(temperature / 100.);
    }