Merge branch 'main' into expr-union-type-impl

Author: chardoncs

Cargo.lock

@@ -20,8 +20,8 @@ pyo3-async-runtimes = { version = "0.24.0", features = ["tokio-runtime"] }
 
 anyhow = { version = "1.0.97", features = ["std"] }
 async-trait = "0.1.88"
-axum = "0.7.9"
-axum-extra = { version = "0.9.6", features = ["query"] }
+axum = "0.8.4"
+axum-extra = { version = "0.10.1", features = ["query"] }
 base64 = "0.22.1"
 chrono = "0.4.40"
 config = "0.14.1"

Cargo.toml

@@ -9,6 +9,8 @@ CocoIndex is an ultra-performant real-time data transformation framework for AI,
 
 As a data framework, CocoIndex takes it to the next level on data freshness. **Incremental processing** is one of the core values provided by CocoIndex.
 
+![Incremental Processing](/img/incremental-etl.gif)
+
 ## Programming Model
 CocoIndex follows the idea of [Dataflow programming](https://en.wikipedia.org/wiki/Dataflow_programming) model. Each transformation creates a new field solely based on input fields, without hidden states and value mutation. All data before/after each transformation is observable, with lineage out of the box.
 

docs/docs/getting_started/overview.md

@@ -2,7 +2,9 @@
 import uvicorn
 from dotenv import load_dotenv
 from fastapi import FastAPI, Query
+from fastapi import Request
 from psycopg_pool import ConnectionPool
+from contextlib import asynccontextmanager
 import os
 
 
@@ -86,27 +88,31 @@ def search(pool: ConnectionPool, query: str, top_k: int = 5):
             ]
 
 
-fastapi_app = FastAPI()
-
-
-@fastapi_app.on_event("startup")
-def startup_event():
+@asynccontextmanager
+def lifespan(app: FastAPI):
     load_dotenv()
     cocoindex.init()
-    # Initialize database connection pool
-    fastapi_app.state.pool = ConnectionPool(os.getenv("COCOINDEX_DATABASE_URL"))
+    pool = ConnectionPool(os.getenv("COCOINDEX_DATABASE_URL"))
+    app.state.pool = pool
+    try:
+        yield
+    finally:
+        pool.close()
+
+
+fastapi_app = FastAPI(lifespan=lifespan)
 
 
 @fastapi_app.get("/search")
 def search_endpoint(
+    request: Request,
     q: str = Query(..., description="Search query"),
     limit: int = Query(5, description="Number of results"),
 ):
-    results = search(fastapi_app.state.pool, q, limit)
+    pool = request.app.state.pool
+    results = search(pool, q, limit)
     return {"results": results}
 
 
 if __name__ == "__main__":
-    load_dotenv()
-    cocoindex.init()
     uvicorn.run(fastapi_app, host="0.0.0.0", port=8080)

docs/static/img/incremental-etl.gif

@@ -1,11 +1,10 @@
 use std::sync::Arc;
 
-use axum::async_trait;
-use futures::{future::BoxFuture, FutureExt};
+use async_trait::async_trait;
+use futures::{FutureExt, future::BoxFuture};
 use pyo3::{
-    pyclass, pymethods,
+    IntoPyObjectExt, Py, PyAny, Python, pyclass, pymethods,
     types::{IntoPyDict, PyString, PyTuple},
-    IntoPyObjectExt, Py, PyAny, Python,
 };
 use pythonize::pythonize;
 
@@ -14,7 +13,7 @@ use crate::{
     builder::plan,
     py::{self, FromPyResult},
 };
-use anyhow::{anyhow, Result};
+use anyhow::{Result, anyhow};
 
 use super::interface::{FlowInstanceContext, SimpleFunctionExecutor, SimpleFunctionFactory};
 

examples/fastapi_server_docker/main.py

@@ -1,7 +1,7 @@
 use crate::prelude::*;
 
 use crate::{lib_context::LibContext, service};
-use axum::{routing, Router};
+use axum::{Router, routing};
 use tower::ServiceBuilder;
 use tower_http::{
     cors::{AllowOrigin, CorsLayer},
@@ -46,31 +46,31 @@ pub async fn init_server(
             Router::new()
                 .route("/flows", routing::get(service::flows::list_flows))
                 .route(
-                    "/flows/:flowInstName",
-                    routing::get(service::flows::get_flow_spec),
+                    "/flows/{flowInstName}",
+                    routing::get(service::flows::get_flow),
                 )
                 .route(
-                    "/flows/:flowInstName/schema",
+                    "/flows/{flowInstName}/schema",
                     routing::get(service::flows::get_flow_schema),
                 )
                 .route(
-                    "/flows/:flowInstName/keys",
+                    "/flows/{flowInstName}/keys",
                     routing::get(service::flows::get_keys),
                 )
                 .route(
-                    "/flows/:flowInstName/data",
+                    "/flows/{flowInstName}/data",
                     routing::get(service::flows::evaluate_data),
                 )
                 .route(
-                    "/flows/:flowInstName/rowStatus",
+                    "/flows/{flowInstName}/rowStatus",
                     routing::get(service::flows::get_row_indexing_status),
                 )
                 .route(
-                    "/flows/:flowInstName/update",
+                    "/flows/{flowInstName}/update",
                     routing::post(service::flows::update),
                 )
                 .route(
-                    "/flows/:flowInstName/search",
+                    "/flows/{flowInstName}/search",
                     routing::get(service::search::search),
                 )
                 .layer(

src/ops/py_factory.rs

@@ -1,14 +1,14 @@
-use anyhow::anyhow;
+use crate::prelude::*;
+
 use axum::{
+    Json,
     http::StatusCode,
     response::{IntoResponse, Response},
 };
-use log::debug;
 use pyo3::{exceptions::PyException, prelude::*};
 use std::{
     error::Error,
     fmt::{Debug, Display},
-    sync::Arc,
 };
 
 #[derive(Debug)]
@@ -38,10 +38,18 @@ impl Error for ApiError {
     }
 }
 
+#[derive(Serialize)]
+struct ErrorResponse {
+    error: String,
+}
+
 impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         debug!("Internal server error:\n{:?}", self.err);
-        (self.status_code, format!("{:#}", self.err)).into_response()
+        let error_response = ErrorResponse {
+            error: self.err.to_string(),
+        };
+        (self.status_code, Json(error_response)).into_response()
     }
 }
 

src/server.rs

@@ -4,9 +4,9 @@ use crate::execution::{evaluator, indexing_status, memoization, row_indexer, sta
 use crate::lib_context::LibContext;
 use crate::{base::schema::FlowSchema, ops::interface::SourceExecutorListOptions};
 use axum::{
+    Json,
     extract::{Path, State},
     http::StatusCode,
-    Json,
 };
 use axum_extra::extract::Query;
 
@@ -18,20 +18,39 @@ pub async fn list_flows(
     ))
 }
 
-pub async fn get_flow_spec(
+pub async fn get_flow_schema(
     Path(flow_name): Path<String>,
     State(lib_context): State<Arc<LibContext>>,
-) -> Result<Json<spec::FlowInstanceSpec>, ApiError> {
+) -> Result<Json<FlowSchema>, ApiError> {
     let flow_ctx = lib_context.get_flow_context(&flow_name)?;
-    Ok(Json(flow_ctx.flow.flow_instance.clone()))
+    Ok(Json(flow_ctx.flow.data_schema.clone()))
 }
 
-pub async fn get_flow_schema(
+#[derive(Serialize)]
+pub struct GetFlowResponse {
+    flow_spec: spec::FlowInstanceSpec,
+    data_schema: FlowSchema,
+    fingerprint: utils::fingerprint::Fingerprint,
+}
+
+pub async fn get_flow(
     Path(flow_name): Path<String>,
     State(lib_context): State<Arc<LibContext>>,
-) -> Result<Json<FlowSchema>, ApiError> {
+) -> Result<Json<GetFlowResponse>, ApiError> {
     let flow_ctx = lib_context.get_flow_context(&flow_name)?;
-    Ok(Json(flow_ctx.flow.data_schema.clone()))
+    let flow_spec = flow_ctx.flow.flow_instance.clone();
+    let data_schema = flow_ctx.flow.data_schema.clone();
+    let fingerprint = utils::fingerprint::Fingerprinter::default()
+        .with(&flow_spec)
+        .map_err(|e| api_error!("failed to fingerprint flow spec: {e}"))?
+        .with(&data_schema)
+        .map_err(|e| api_error!("failed to fingerprint data schema: {e}"))?
+        .into_fingerprint();
+    Ok(Json(GetFlowResponse {
+        flow_spec,
+        data_schema,
+        fingerprint,
+    }))
 }
 
 #[derive(Deserialize)]