Apache Doris connector

use cocoindex::connectors::doris::{
    self, ColumnDef, DorisConfig, DorisConnection, DorisTableOptions, InvertedIndexDef, TableSchema,
};

let mut cfg = DorisConfig::new(fe_host, database)
    .fe_http_port(http_port)
    .query_port(query_port);
if let Ok(user) = std::env::var("DORIS_USERNAME") {
    cfg = cfg.username(user);
}
if let Ok(pw) = std::env::var("DORIS_PASSWORD") {
    cfg = cfg.password(pw);
}

impl DorisConnection {
    pub async fn connect(config: DorisConfig) -> Result<Self>
    pub fn pool(&self) -> &MySqlPool
    pub fn config(&self) -> &DorisConfig
}

let conn = DorisConnection::connect(cfg).await?;

use cocoindex::connectors::doris::{DorisConfig, DorisConnection};

cocoindex::context_key!(static DORIS_DB: DorisConnection, key = "doris_target_test_db");

let conn = DorisConnection::connect(cfg).await?;
let app = Environment::builder()
    .db_path(&db)
    .provide_key(&DORIS_DB, conn.clone())
    .build()
    .await?
    .app("DorisInvertedIndex")
    .await?;

pub async fn mount_table_target(
    ctx: &Ctx,
    conn: &ContextKey<DorisConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
) -> Result<TableTarget>

pub async fn mount_table_target_with_options(
    ctx: &Ctx,
    conn: &ContextKey<DorisConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    options: DorisTableOptions,
) -> Result<TableTarget>

pub struct DorisTableOptions {
    pub managed_by: ManagedBy,
    pub vector_indexes: Vec<VectorIndexDef>,
    pub inverted_indexes: Vec<InvertedIndexDef>,
}

#[cocoindex::function]
async fn process_facts(ctx: &Ctx, file: FileEntry, table: doris::TableTarget) -> Result<()> {
    // ...
}

mount_each!("facts", facts, |file| process_facts(ctx, file, table)).await?;

impl TableTarget {
    pub fn declare_row<R: Serialize>(&self, ctx: &Ctx, row: &R) -> Result<()>
}

target.declare_row(
    &ctx,
    &serde_json::json!({"id": "1", "content": "the quick brown fox"}),
)?;

impl TableSchema {
    pub fn from_row<T: SchemaFields>(
        primary_key: impl IntoIterator<Item = impl Into<String>>,
    ) -> Result<Self>

    pub fn with_vector_dim(self, field_name: &str, dim: usize) -> Result<Self>
    pub async fn with_vector_schema(
        self,
        field_name: &str,
        provider: &(impl VectorSchemaProvider + ?Sized),
    ) -> Result<Self>
    pub fn with_column_type(self, field_name: &str, doris_type: impl Into<String>) -> Result<Self>
}

#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct DocEmbedding {
    id: i64,
    text: String,
    #[coco(vector)]
    embedding: Vec<f32>,
}

let schema = doris::TableSchema::from_row::<DocEmbedding>(["id"])?
    .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
    .await?;

#[derive(Serialize, SchemaFields)]
struct MyRow {
    #[coco(type = "INT")]
    id: i64,
    #[coco(type = "FLOAT")]
    value: f64,
}

pub struct VectorIndexDef {
    pub field_name: String,
    pub index_type: String,   // "HNSW" (default) or "IVF"
    pub metric_type: String,  // "l2_distance" (default), "inner_product", "cosine_distance"
    pub max_degree: Option<u32>,
    pub ef_construction: Option<u32>,
    pub nlist: Option<u32>,
}

impl VectorIndexDef {
    pub fn new(field_name: impl Into<String>) -> Self
    pub fn index_type(self, index_type: impl Into<String>) -> Self
    pub fn metric_type(self, metric: impl Into<String>) -> Self
    pub fn max_degree(self, v: u32) -> Self
    pub fn ef_construction(self, v: u32) -> Self
    pub fn nlist(self, v: u32) -> Self
}

let options = DorisTableOptions {
    inverted_indexes: vec![InvertedIndexDef::new("content").parser("english")],
    ..Default::default()
};

let target =
    doris::mount_table_target_with_options(&ctx, &DORIS_DB, table, schema, options)
        .await?;

pub fn build_vector_search_query(
    table: &str,
    vector_field: &str,
    query_vector: &[f32],
    options: &VectorSearchOptions,
) -> Result<String>

pub struct VectorSearchOptions {
    pub metric: String,               // default "l2_distance"
    pub limit: usize,                 // default 10
    pub select_columns: Vec<String>,  // empty selects `*`
    pub where_clause: Option<String>,
}

let sql = doris::build_vector_search_query(
    "doc_embeddings",
    "embedding",
    &query_vector,
    &doris::VectorSearchOptions {
        select_columns: vec!["id".into(), "text".into()],
        ..Default::default()
    },
)?;

use cocoindex::connectors::doris::{
    self, DorisConfig, DorisConnection, DorisTableOptions, VectorIndexDef,
};
use cocoindex::prelude::*;

cocoindex::context_key!(static DORIS_DB: DorisConnection, key = "doris_target_test_db");

#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct DocEmbedding {
    id: i64,
    text: String,
    #[coco(vector = 384)]
    embedding: Vec<f32>,
}

let conn = DorisConnection::connect(DorisConfig::new("localhost", "my_database")).await?;
let app = Environment::builder()
    .db_path(&db)
    .provide_key(&DORIS_DB, conn.clone())
    .build()
    .await?
    .app("DorisExample")
    .await?;

app.run(move |ctx| async move {
    let options = DorisTableOptions {
        vector_indexes: vec![
            VectorIndexDef::new("embedding")
                .index_type("HNSW")
                .metric_type("cosine_distance"),
        ],
        ..Default::default()
    };
    let target = doris::mount_table_target_with_options(
        &ctx,
        &DORIS_DB,
        "doc_embeddings",
        doris::TableSchema::from_row::<DocEmbedding>(["id"])?,
        options,
    )
    .await?;

    // Declare rows
    for doc in &documents {
        target.declare_row(&ctx, doc)?;
    }
    Ok(())
})
.await?;