versioned datasets v0 (#999)

* versioned datasets v0

* fix build

* order table by func on id, drop additional sort from views

* update get datapoints on app server, update view name

* fix view, accept id in post datapoints

Author: dinmukhamedm

app-server/Cargo.toml

@@ -56,7 +56,7 @@ tracing = {version = "0.1.41", features = ["attributes"]}
 tracing-opentelemetry = "0.30.0"
 tracing-subscriber = {version = "0.3", features = ["env-filter", "fmt"]}
 url = "2.5.4"
-uuid = {version = "1.18.1", features = ["v4", "fast-rng", "macro-diagnostics", "serde"]}
+uuid = {version = "1.18.1", features = ["v4", "fast-rng", "macro-diagnostics", "serde", "v7", "std"]}
 
 [build-dependencies]
 tonic-build = "0.13"

app-server/src/api/v1/datasets.rs

@@ -1,21 +1,27 @@
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
 
 use actix_web::{HttpResponse, get, post, web};
+use chrono::Utc;
 use futures_util::StreamExt;
 use serde::Deserialize;
+use serde_json::Value;
 use uuid::Uuid;
 
 use crate::{
-    ch::datapoints as ch_datapoints,
-    datasets::datapoints::Datapoint,
+    ch::datapoints::{self as ch_datapoints},
+    datasets::datapoints::{CHQueryEngineDatapoint, Datapoint},
     db::{self, DB, project_api_keys::ProjectApiKey},
+    query_engine::QueryEngine,
     routes::{PaginatedResponse, types::ResponseResult},
+    sql::{self, ClickhouseReadonlyClient},
     storage::{Storage, StorageTrait},
 };
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct GetDatapointsRequestParams {
-    name: String,
+    #[serde(alias = "dataset_name", alias = "name")]
+    dataset_name: String,
     limit: i64,
     offset: i64,
 }
@@ -24,39 +30,93 @@ pub struct GetDatapointsRequestParams {
 async fn get_datapoints(
     params: web::Query<GetDatapointsRequestParams>,
     db: web::Data<DB>,
-    clickhouse: web::Data<clickhouse::Client>,
+    clickhouse_ro: web::Data<Option<Arc<ClickhouseReadonlyClient>>>,
+    query_engine: web::Data<Arc<QueryEngine>>,
     project_api_key: ProjectApiKey,
 ) -> ResponseResult {
     let project_id = project_api_key.project_id;
     let db = db.into_inner();
-    let clickhouse = clickhouse.into_inner().as_ref().clone();
+    let clickhouse_ro = if let Some(clickhouse_ro) = clickhouse_ro.as_ref() {
+        clickhouse_ro.clone()
+    } else {
+        return Ok(HttpResponse::InternalServerError().json(serde_json::json!({
+            "error": "ClickHouse read-only client is not configured"
+        })));
+    };
+    let query_engine = query_engine.into_inner().as_ref().clone();
     let query = params.into_inner();
 
     let dataset_id =
-        db::datasets::get_dataset_id_by_name(&db.pool, &query.name, project_id).await?;
+        db::datasets::get_dataset_id_by_name(&db.pool, &query.dataset_name, project_id).await?;
 
     let Some(dataset_id) = dataset_id else {
         return Ok(HttpResponse::NotFound().json(serde_json::json!({
             "error": "Dataset not found"
         })));
     };
 
-    // Get datapoints from ClickHouse
-    let ch_datapoints = ch_datapoints::get_datapoints_paginated(
-        clickhouse.clone(),
+    let select_query = "
+        SELECT
+            id,
+            dataset_id,
+            created_at,
+            data,
+            target,
+            metadata
+        FROM dataset_datapoints
+        WHERE dataset_id = {dataset_id:UUID}
+        ORDER BY toUInt128(id) ASC
+        LIMIT {limit:Int64}
+        OFFSET {offset:Int64}
+    ";
+    let parameters = HashMap::from([
+        (
+            "dataset_id".to_string(),
+            Value::String(dataset_id.to_string()),
+        ),
+        ("limit".to_string(), Value::Number(query.limit.into())),
+        ("offset".to_string(), Value::Number(query.offset.into())),
+    ]);
+
+    let select_query_result = sql::execute_sql_query(
+        select_query.to_string(),
         project_id,
-        dataset_id,
-        Some(query.limit),
-        Some(query.offset),
+        parameters.clone(),
+        clickhouse_ro.clone(),
+        query_engine.clone(),
     )
     .await?;
 
-    let total_count = ch_datapoints::count_datapoints(clickhouse, project_id, dataset_id).await?;
+    let total_count_query = "
+        SELECT COUNT(*) as count FROM dataset_datapoints
+        WHERE dataset_id = {dataset_id:UUID}
+    ";
 
-    let datapoints: Vec<Datapoint> = ch_datapoints
+    let total_count_result = sql::execute_sql_query(
+        total_count_query.to_string(),
+        project_id,
+        HashMap::from([(
+            "dataset_id".to_string(),
+            Value::String(dataset_id.to_string()),
+        )]),
+        clickhouse_ro,
+        query_engine,
+    )
+    .await?;
+
+    let total_count = total_count_result
+        .first()
+        .and_then(|v| v.get("count").and_then(|v| v.as_i64()).map(|v| v as u64))
+        .unwrap_or_default();
+
+    let datapoints: Vec<Datapoint> = select_query_result
         .into_iter()
-        .map(|ch_dp| ch_dp.into())
-        .collect();
+        .map(|ch_dp| {
+            serde_json::from_value::<CHQueryEngineDatapoint>(ch_dp)
+                .map_err(anyhow::Error::from)
+                .and_then(|ch_dp| ch_dp.try_into())
+        })
+        .collect::<Result<Vec<Datapoint>, anyhow::Error>>()?;
 
     let response = PaginatedResponse {
         total_count,
@@ -68,13 +128,18 @@ async fn get_datapoints(
 }
 
 #[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct CreateDatapointsRequest {
+    // The alias is added to support the old endpoint (dataset_name)
+    #[serde(alias = "dataset_name")]
     pub dataset_name: String,
     pub datapoints: Vec<CreateDatapointRequest>,
 }
 
 #[derive(Deserialize)]
 pub struct CreateDatapointRequest {
+    #[serde(default)]
+    pub id: Option<Uuid>,
     pub data: serde_json::Value,
     pub target: Option<serde_json::Value>,
     #[serde(default)]
@@ -119,7 +184,9 @@ async fn create_datapoints(
         .datapoints
         .into_iter()
         .map(|dp_req| Datapoint {
-            id: Uuid::new_v4(),
+            // now_v7 is guaranteed to be sorted by creation time
+            id: dp_req.id.unwrap_or(Uuid::now_v7()),
+            created_at: Utc::now(),
             dataset_id,
             data: dp_req.data,
             target: dp_req.target,

app-server/src/ch/datapoints.rs

@@ -1,8 +1,7 @@
 use anyhow::Result;
-use chrono::Utc;
+use chrono::{DateTime, Utc};
 use clickhouse::Row;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
 use uuid::Uuid;
 
 use super::utils::chrono_to_nanoseconds;
@@ -27,18 +26,17 @@ pub struct CHDatapoint {
 impl From<CHDatapoint> for Datapoint {
     fn from(ch_datapoint: CHDatapoint) -> Self {
         // Parse JSON strings back to Values
-        let data = serde_json::from_str(&ch_datapoint.data).unwrap_or(serde_json::Value::Null);
-        let target = if ch_datapoint.target == "<null>" || ch_datapoint.target.is_empty() {
-            None
-        } else {
-            serde_json::from_str(&ch_datapoint.target).ok()
-        };
-        let metadata: HashMap<String, serde_json::Value> =
-            serde_json::from_str(&ch_datapoint.metadata).unwrap_or_default();
+        let (data, target, metadata) = Datapoint::parse_string_payloads(
+            ch_datapoint.data,
+            ch_datapoint.target,
+            ch_datapoint.metadata,
+        )
+        .unwrap();
 
         Datapoint {
             id: ch_datapoint.id,
             dataset_id: ch_datapoint.dataset_id,
+            created_at: DateTime::from_timestamp_nanos(ch_datapoint.created_at),
             data,
             target,
             metadata,
@@ -99,63 +97,3 @@ pub async fn insert_datapoints(
         )),
     }
 }
-
-/// Get paginated datapoints from ClickHouse
-pub async fn get_datapoints_paginated(
-    clickhouse: clickhouse::Client,
-    project_id: Uuid,
-    dataset_id: Uuid,
-    limit: Option<i64>,
-    offset: Option<i64>,
-) -> Result<Vec<CHDatapoint>> {
-    let mut query = String::from(
-        "SELECT 
-            id,
-            dataset_id,
-            project_id,
-            created_at,
-            data,
-            target,
-            metadata
-        FROM dataset_datapoints
-        WHERE project_id = ? AND dataset_id = ?
-        ORDER BY created_at DESC",
-    );
-
-    if let Some(limit) = limit {
-        query.push_str(&format!(" LIMIT {}", limit));
-    }
-    if let Some(offset) = offset {
-        query.push_str(&format!(" OFFSET {}", offset));
-    }
-
-    let datapoints = clickhouse
-        .query(&query)
-        .bind(project_id)
-        .bind(dataset_id)
-        .fetch_all::<CHDatapoint>()
-        .await?;
-
-    Ok(datapoints)
-}
-
-#[derive(Row, Deserialize)]
-struct CountResult {
-    count: u64,
-}
-
-/// Count total datapoints in ClickHouse
-pub async fn count_datapoints(
-    clickhouse: clickhouse::Client,
-    project_id: Uuid,
-    dataset_id: Uuid,
-) -> Result<u64> {
-    let result = clickhouse
-        .query("SELECT COUNT(*) as count FROM dataset_datapoints WHERE project_id = ? AND dataset_id = ?")
-        .bind(project_id)
-        .bind(dataset_id)
-        .fetch_one::<CountResult>()
-        .await?;
-
-    Ok(result.count)
-}

app-server/src/datasets/datapoints.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 
-use serde::Serialize;
+use chrono::{DateTime, NaiveDateTime, Utc};
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use uuid::Uuid;
 
@@ -12,4 +13,56 @@ pub struct Datapoint {
     pub data: Value,
     pub target: Option<Value>,
     pub metadata: HashMap<String, Value>,
+    pub created_at: DateTime<Utc>,
+}
+
+impl Datapoint {
+    pub fn parse_string_payloads(
+        data: String,
+        target: String,
+        metadata: String,
+    ) -> Result<(Value, Option<Value>, HashMap<String, Value>), serde_json::Error> {
+        let data = serde_json::from_str(&data).unwrap_or(serde_json::Value::Null);
+        let target = if target == "<null>" || target.is_empty() || target.to_lowercase() == "null" {
+            None
+        } else {
+            serde_json::from_str(&target).ok()
+        };
+        let metadata: HashMap<String, serde_json::Value> =
+            serde_json::from_str(&metadata).unwrap_or_default();
+
+        Ok((data, target, metadata))
+    }
+}
+
+// This struct is similar to CHDatapoint, but for some reason
+// uuid parsing on that doesn't work with the query engine
+// returned owned strings (internally it uses Uuid::from_str which expects a &str)
+#[derive(Deserialize)]
+pub struct CHQueryEngineDatapoint {
+    pub id: Uuid,
+    pub dataset_id: Uuid,
+    pub created_at: String,
+    pub data: String,
+    pub target: String,
+    pub metadata: String,
+}
+
+impl TryInto<Datapoint> for CHQueryEngineDatapoint {
+    type Error = anyhow::Error;
+    fn try_into(self) -> Result<Datapoint, Self::Error> {
+        let (data, target, metadata) =
+            Datapoint::parse_string_payloads(self.data, self.target, self.metadata)?;
+        Ok(Datapoint {
+            id: self.id,
+            dataset_id: self.dataset_id,
+            created_at: DateTime::from_naive_utc_and_offset(
+                NaiveDateTime::parse_from_str(&self.created_at, "%Y-%m-%d %H:%M:%S%.f")?,
+                Utc,
+            ),
+            data,
+            target,
+            metadata,
+        })
+    }
 }

frontend/app/api/projects/[projectId]/datasets/[datasetId]/count/route.ts

@@ -0,0 +1,22 @@
+import { NextRequest, NextResponse } from "next/server";
+
+import { countDatapoints } from "@/lib/actions/datapoints";
+
+export async function GET(
+  req: NextRequest,
+  props: { params: Promise<{ projectId: string; datasetId: string }> }
+): Promise<Response> {
+  const params = await props.params;
+
+  try {
+    const countData = await countDatapoints({
+      projectId: params.projectId,
+      datasetId: params.datasetId,
+    });
+
+    return NextResponse.json(countData);
+  } catch (error) {
+    console.error("Error listing datapoints:", error);
+    return NextResponse.json({ error: "Internal server error" }, { status: 500 });
+  }
+}