Merge remote-tracking branch 'origin/develop' into ji/min-max-rt

# Conflicts:
#	vortex-array/src/arrays/decimal/compute/min_max.rs

Author: joseph-isaacs

benchmarks-website/code.js

@@ -679,38 +679,73 @@ window.initAndRender = (function () {
   const initializer = {
     async loadData() {
       const [dataResponse, commitsResponse] = await Promise.all([
-        this.fetchGzippedData(
+        this.fetchAndParseGzippedJsonl(
           "https://vortex-benchmark-results-database.s3.amazonaws.com/data.json.gz"
         ),
         fetch(
           "https://vortex-benchmark-results-database.s3.amazonaws.com/commits.json"
         ).then((r) => r.text()),
       ]);
 
-      // Return raw text data for worker processing
+      // Return parsed data for worker processing
+      // dataResponse is now an array of parsed objects instead of a string
       return {
         benchmarkData: dataResponse,
         commitsData: commitsResponse
       };
     },
 
-    async fetchGzippedData(url) {
+    async fetchAndParseGzippedJsonl(url) {
       const response = await fetch(url);
       const decompressedStream = response.body.pipeThrough(
         new DecompressionStream("gzip")
       );
       const reader = decompressedStream.getReader();
       const decoder = new TextDecoder();
-      let result = "";
+
+      let buffer = '';
+      const lines = [];
 
       while (true) {
         const { done, value } = await reader.read();
-        if (done) break;
-        result += decoder.decode(value, { stream: true });
+
+        if (value) {
+          buffer += decoder.decode(value, { stream: true });
+
+          // Process complete lines
+          const parts = buffer.split('\n');
+          // Keep the last part (incomplete line) in the buffer
+          buffer = parts.pop();
+
+          // Parse and collect complete lines
+          for (const line of parts) {
+            const trimmed = line.trim();
+            if (trimmed.length > 0) {
+              try {
+                lines.push(JSON.parse(trimmed));
+              } catch (e) {
+                console.warn('Failed to parse JSONL line:', trimmed, e);
+              }
+            }
+          }
+        }
+
+        if (done) {
+          // Process any remaining buffer
+          buffer += decoder.decode();
+          const trimmed = buffer.trim();
+          if (trimmed.length > 0) {
+            try {
+              lines.push(JSON.parse(trimmed));
+            } catch (e) {
+              console.warn('Failed to parse final JSONL line:', trimmed, e);
+            }
+          }
+          break;
+        }
       }
 
-      result += decoder.decode();
-      return result;
+      return lines;
     },
 
     parseJsonl(jsonl) {
@@ -754,7 +789,7 @@ window.initAndRender = (function () {
       ];
 
       elementIds.forEach((id) => {
-        const camelCaseId = id.replace(/-(.)/g, (match, char) =>
+        const camelCaseId = id.replace(/-(.)/g, (_match, char) =>
           char.toUpperCase()
         );
         domElements[camelCaseId] = document.getElementById(id);

benchmarks-website/data-worker.js

@@ -323,7 +323,10 @@ self.addEventListener('message', async function(e) {
           message: `${getRandomMessage(PARSING_MESSAGES)} Benchmark data...`
         });
 
-        const benchmarkData = parseJsonl(data.benchmarkData);
+        // benchmarkData might already be parsed as an array
+        const benchmarkData = Array.isArray(data.benchmarkData)
+          ? data.benchmarkData
+          : parseJsonl(data.benchmarkData);
 
         self.postMessage({
           type: 'progress',

benchmarks-website/worker-manager.js

@@ -89,16 +89,18 @@ export const workerManager = {
   // Fallback data processing on main thread
   async processDataFallback(benchmarkData, commitsData, keptGroups, onProgress) {
     if (onProgress) onProgress(10, 'Parsing benchmark data...');
-    
-    // Parse JSONL data
-    const parsedBenchmarkData = this.parseJsonl(benchmarkData);
-    
+
+    // benchmarkData is now already parsed as an array
+    const parsedBenchmarkData = Array.isArray(benchmarkData)
+      ? benchmarkData
+      : this.parseJsonl(benchmarkData);
+
     if (onProgress) onProgress(30, 'Parsing commit data...');
-    
+
     const parsedCommitsData = this.parseJsonl(commitsData);
-    
+
     if (onProgress) onProgress(50, 'Processing and grouping data...');
-    
+
     // Convert commits array to object
     const commits = {};
     parsedCommitsData.forEach((commit) => {
@@ -113,7 +115,7 @@ export const workerManager = {
     );
 
     if (onProgress) onProgress(100, 'Data processing complete!');
-    
+
     return result;
   },
 

encodings/alp/benches/alp_compress.rs

@@ -6,11 +6,11 @@
 use divan::Bencher;
 use rand::rngs::StdRng;
 use rand::{Rng, SeedableRng as _};
-use vortex_alp::{ALPFloat, ALPRDFloat, RDEncoder, alp_encode};
+use vortex_alp::{ALPFloat, ALPRDFloat, RDEncoder, alp_encode, decompress};
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::compute::warm_up_vtables;
 use vortex_array::validity::Validity;
-use vortex_buffer::buffer;
+use vortex_buffer::{Buffer, buffer};
 use vortex_dtype::NativePType;
 
 fn main() {
@@ -84,10 +84,15 @@ fn decompress_alp<T: ALPFloat + NativePType>(bencher: Bencher, args: (usize, f64
         Validity::NonNullable
     };
     let values = values.freeze();
-    let array = alp_encode(&PrimitiveArray::new(values, validity), None).unwrap();
     bencher
-        .with_inputs(|| array.clone())
-        .bench_values(|array| array.to_canonical());
+        .with_inputs(|| {
+            alp_encode(
+                &PrimitiveArray::new(Buffer::copy_from(&values), validity.clone()),
+                None,
+            )
+            .unwrap()
+        })
+        .bench_values(decompress);
 }
 
 #[divan::bench(types = [f32, f64], args = [10_000, 100_000])]

encodings/alp/src/alp/array.rs

@@ -282,6 +282,6 @@ impl ArrayVTable<ALPVTable> for ALPVTable {
 
 impl CanonicalVTable<ALPVTable> for ALPVTable {
     fn canonicalize(array: &ALPArray) -> Canonical {
-        Canonical::Primitive(decompress(array))
+        Canonical::Primitive(decompress(array.clone()))
     }
 }