Merge branch 'master' into fix/empty_labels_check

Author: lmangani

.github/workflows/node-clickhouse.js.yml

@@ -26,7 +26,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        node-version: [18, 16, 20]
+        node-version: [18, 20, 22]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
     services:
       clickhouse:

.gitignore

@@ -9,3 +9,6 @@ node_modules
 /wasm_parts/vendor/
 /wasm_parts/main.wasm
 /wasm_parts/wasm_parts.iml
+/test/qryn_test_env/clickhouse/_data/
+/test/qryn_test_env/grafana/_data/
+/test/qryn_test_cluster_env/grafana/_data/

Dockerfile

@@ -1,5 +1,5 @@
 # qryn
-FROM node:20.10-slim
+FROM node:20.17-slim
 
 COPY . /app
 WORKDIR /app

lib/db/maintain/scripts.js

@@ -198,7 +198,7 @@ module.exports.traces = [
 ]
 
 module.exports.overall_dist = [
-  `CREATE TABLE {{DB}}.metrics_15s_dist {{{OnCluster}}} (
+  `CREATE TABLE IF NOT EXISTS {{DB}}.metrics_15s_dist {{{OnCluster}}} (
     \`fingerprint\` UInt64,
     \`timestamp_ns\` Int64 CODEC(DoubleDelta),
     \`last\` AggregateFunction(argMax, Float64, Int64),

lib/handlers/push.js

@@ -71,7 +71,7 @@ function processStream (stream, labels, bulkLabels, bulk, toJSON, fingerPrint) {
       values.push([
         finger,
         ts,
-        (typeof entry.value === 'undefined') ? null : entry.value,
+        (typeof entry.value !== 'number') ? 0 : entry.value,
         entry.line || '',
         type === 3 ? bothType : type
       ])
@@ -98,7 +98,7 @@ function processStream (stream, labels, bulkLabels, bulk, toJSON, fingerPrint) {
       values.push([
         finger,
         BigInt(value[0]),
-        (typeof value[2] === 'undefined') ? null : value[2],
+        (typeof value[2] !== 'number') ? 0 : value[2],
         value[1] || '',
         type === 3 ? bothType : type
       ])

package-lock.json

@@ -445,7 +445,7 @@ module.exports.preJoinLabels = (token, query, dist) => {
   dist = dist || ''
   const timeSeriesReq = new Sql.Select()
     .select('fingerprint', 'labels')
-    .from([`${DATABASE_NAME()}.time_series${dist}`, 'time_series'])
+    .from([`${DATABASE_NAME()}.time_series`, 'time_series'])
     .where(new Sql.And(
       new Sql.In('time_series.fingerprint', 'in', inRightSide),
       Sql.Gte(new Sql.Raw('date'), sqlFrom),

parser/registry/common.js

@@ -1,4 +1,4 @@
-const { getDuration, preJoinLabels, dist } = require('../common')
+const { getDuration, preJoinLabels, dist, sharedParamNames } = require('../common')
 const reg = require('./log_range_agg_reg_v3_2')
 const Sql = require('@cloki/clickhouse-sql')
 const { DATABASE_NAME, checkVersion } = require('../../../lib/utils')
@@ -44,24 +44,30 @@ function isLogPipeline (token) {
  */
 module.exports.apply = (token, fromNS, toNS, stepNS) => {
   fromNS = Math.floor(fromNS / 15000000000) * 15000000000
+  const fromParam = new Sql.Parameter(sharedParamNames.from)
+  const toParam = new Sql.Parameter(sharedParamNames.to)
   const tsClause = toNS
     ? Sql.between('samples.timestamp_ns', fromNS, toNS)
     : Sql.Gt('samples.timestamp_ns', fromNS)
   let q = (new Sql.Select())
     .select(['samples.fingerprint', 'fingerprint'])
     .from([`${DATABASE_NAME()}.metrics_15s${_dist}`, 'samples'])
     .where(tsClause)
+    .addParam(fromParam)
+    .addParam(toParam)
+  fromParam.set(fromNS)
+  toParam.set(toNS)
 
   q.ctx = {
     step: stepNS / 1000000000,
     inline: !!clusterName
   }
 
-  preJoinLabels(token, q, dist)
-
   for (const streamSelectorRule of token.Children('log_stream_selector_rule')) {
     q = streamSelectorReg[streamSelectorRule.Child('operator').value](streamSelectorRule, q)
   }
+  preJoinLabels(token, q, dist)
+  q = q.groupBy('labels')
 
   const lra = token.Child('log_range_aggregation')
   q = reg[lra.Child('log_range_aggregation_fn').value](lra, q)

parser/registry/smart_optimizations/optimization_v3_2.js

@@ -60,7 +60,9 @@ module.exports.series = async (query, fromMs, toMs) => {
     const data = await rawRequest(req.toString() + ' FORMAT JSON',
       null,
       DATABASE_NAME())
-    return data.data.data.map(l => JSON.parse(l.labels))
+    return data.data.data.map(l =>
+      Object.fromEntries(Object.entries(JSON.parse(l.labels)).filter(e => e[1]))
+    )
   } catch (e) {
     if (e instanceof prometheus.WasmError) {
       throw new PSQLError(e.message)

promql/index.js

@@ -27,9 +27,15 @@ flate2 = "1.0"
 # code size when deploying.
 console_error_panic_hook = { version = "0.1.7", optional = true }
 base64 = "0.22.1"
+memchr = "2.7.4"
 
 [dev-dependencies]
 wasm-bindgen-test = "0.3.34"
+criterion = { version = "0.5.1", features = ["html_reports"] }
+
+[[bench]]
+name = "my_benchmark"
+harness = false
 
 [profile.release]
 # Tell `rustc` to optimize for small code size.

pyroscope/pprof-bin/Cargo.toml

@@ -0,0 +1,18 @@
+use pprof_bin::merge_prof;
+use pprof_bin::utest::get_test_pprof_data;
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+
+fn merge_bench(pprofs: &Vec<Vec<u8>>) {
+
+    for pprof in pprofs {
+        merge_prof(0, pprof.as_slice(), "process_cpu:samples:count:cpu:nanoseconds".to_string());
+    }
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let pprofs = get_test_pprof_data();
+    c.bench_function("merge", |b| b.iter(|| merge_bench(&pprofs)));
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);