1brc challange submission jatingala (#364)

* add code

* enable parallel

* fix code warnings

* use graal vm

* formatting changes by build

* add license

1 files changed, 154 insertions, 0 deletions

diff --git a/src/main/java/dev/morling/onebrc/CalculateAverage_jatingala.java b/src/main/java/dev/morling/onebrc/CalculateAverage_jatingala.java
new file mode 100644
index 0000000..8eca51c
--- /dev/null
+++ b/src/main/java/dev/morling/onebrc/CalculateAverage_jatingala.java
@@ -0,0 +1,154 @@
+/*
+ *  Copyright 2023 The original authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package dev.morling.onebrc;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.MappedByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.*;
+
+public class CalculateAverage_jatingala {
+
+    private static final String FILE = "./measurements.txt";
+
+    public static void main(final String[] args) throws IOException {
+        final int processorCount = Math.max(Runtime.getRuntime().availableProcessors(), 8);
+
+        try (final RandomAccessFile randomAccessFile = new RandomAccessFile(FILE, "r")) {
+            final long[][] chunks = getChunkPositions(randomAccessFile, processorCount);
+            final Map<String, Statistics> result = Arrays.stream(chunks)
+                    .parallel()
+                    .map(chunk -> {
+                        try {
+                            final MappedByteBuffer mappedByteBuffer = randomAccessFile.getChannel().map(FileChannel.MapMode.READ_ONLY, chunk[0], chunk[1]);
+                            return consumeChunk(mappedByteBuffer);
+                        }
+                        catch (final Exception e) {
+                            System.out.println(e.getMessage());
+                            return new HashMap<String, Statistics>();
+                        }
+                    })
+                    .reduce((a, b) -> {
+                        a.forEach((k, v) -> {
+                            final Statistics s = b.get(k);
+                            if (s != null)
+                                v.merge(s);
+                        });
+                        b.forEach(a::putIfAbsent);
+                        return a;
+                    })
+                    .orElseGet(Collections::emptyMap);
+
+            System.out.println(new TreeMap<>(result));
+        }
+    }
+
+    private static long[][] getChunkPositions(final RandomAccessFile file, final int chunks) throws IOException {
+        final long[][] result = new long[chunks][];
+
+        final long fileSize = file.length();
+        final long chunkSize = Math.ceilDiv(fileSize, chunks);
+        long chunkStartPosition = 0;
+        for (int i = 0; i < chunks; i++) {
+            file.seek(Math.min(chunkStartPosition + chunkSize, fileSize));
+
+            while (file.getFilePointer() < fileSize && file.readByte() != '\n') {
+                // find next newline, noop
+            }
+
+            // startPointer & length
+            result[i] = new long[]{ chunkStartPosition, file.getFilePointer() - chunkStartPosition };
+            chunkStartPosition = file.getFilePointer();
+        }
+
+        return result;
+    }
+
+    private static Map<String, Statistics> consumeChunk(final MappedByteBuffer mappedByteBuffer) {
+        final Map<String, Statistics> statisticsMap = new HashMap<>();
+
+        while (mappedByteBuffer.hasRemaining()) {
+            final String key = parseKey(mappedByteBuffer);
+            final double value = parseNumber(mappedByteBuffer);
+            statisticsMap.computeIfAbsent(key, _ -> new Statistics()).update(value);
+        }
+
+        return statisticsMap;
+    }
+
+    private static String parseKey(final MappedByteBuffer mappedByteBuffer) {
+        final ByteArrayOutputStream keyBytes = new ByteArrayOutputStream();
+        while (mappedByteBuffer.hasRemaining()) {
+            final byte val = mappedByteBuffer.get();
+            if (val == ';')
+                break;
+            keyBytes.write(val);
+        }
+        return keyBytes.toString();
+    }
+
+    private static double parseNumber(final MappedByteBuffer mappedByteBuffer) {
+        boolean negate = false;
+        int temp = 0;
+        while (mappedByteBuffer.hasRemaining()) {
+            final byte val = mappedByteBuffer.get();
+
+            if (val == '\n')
+                break;
+            if (val == '-') {
+                negate = true;
+                continue;
+            }
+            if (val == '.')
+                continue;
+
+            temp = 10 * temp + (val - '0');
+        }
+        return (negate ? -1 : 1) * (temp / 10.0);
+    }
+
+    private static class Statistics {
+        private double min = Double.POSITIVE_INFINITY;
+        private double max = Double.NEGATIVE_INFINITY;
+        private double sum;
+        private long count;
+
+        private static double round(final double value) {
+            return Math.round(value * 10.0) / 10.0;
+        }
+
+        public void update(final double reading) {
+            this.min = Math.min(min, reading);
+            this.max = Math.max(max, reading);
+            this.sum += reading;
+            ++count;
+        }
+
+        public void merge(final Statistics other) {
+            this.min = Math.min(min, other.min);
+            this.max = Math.max(max, other.max);
+            this.sum += other.sum;
+            this.count += other.count;
+        }
+
+        @Override
+        public String toString() {
+            return STR."\{round(min)}/\{round(sum / count)}/\{round(max)}";
+        }
+    }
+}