use sea_query::Alias;

use sea_query::{Expr,NullOrdering,Order,PostgresQueryBuilder,Query};

use sea_query_binder::SqlxBinder;

use serde_json::json;

use sqlx::Executor;

use serde_json::{json,Value};

use sqlx::PgConnection;

use sqlx::{Executor,Pool,Postgres};

use std::borrow::Cow;

use std::collections::HashMap;

use std::path::Path;

use std::time::SystemTime;

use std::time::UNIX_EPOCH;

use tokio::task::JoinSet;

use tracing::{instrument, warn};

use walkdir::WalkDir;

usecrate::debug_sqlx_query;

usecrate::filter_builder::FilterBuilder;

usecrate::pipeline::FieldAction;

usecrate::search_query_builder::build_search_query;

usecrate::vector_search_query_builder::build_vector_search_query;

letmut args = args.unwrap_or_default();

let args = args.as_object_mut().context("args must be a JSON object")?;

self.verify_in_database(false).await?;

letmut pipelines =self.get_pipelines().await?;

let pool =get_or_initialize_pool(&self.database_url).await?;

letmut parsed_schemas =vec![];

let project_info =&self.database_data.as_ref().unwrap().project_info;

letmut parsed_schemas =vec![];

for pipelinein&mut pipelines{

let parsed_schema = pipeline

.get_parsed_schema(project_info,&pool)

parsed_schemas.push(parsed_schema);

letmut pipelines:Vec<(Pipeline,_)> = pipelines.into_iter().zip(parsed_schemas).collect();

let pipelines:Vec<(Pipeline,HashMap<String,FieldAction>)> =

pipelines.into_iter().zip(parsed_schemas).collect();

let args = args.unwrap_or_default();

let args = args.as_object().context("args must be a JSON object")?;

let batch_size = args

.map(|x| x.try_to_u64())

.unwrap_or(Ok(100))?;

let parallel_batches = args

.map(|x| x.try_to_u64())

.unwrap_or(Ok(1))?asusize;

let progress_bar = utils::default_progress_bar(documents.len()asu64);

progress_bar.println("Upserting Documents...");

letmut set =JoinSet::new();

for batchin documents.chunks(batch_sizeasusize){

if set.len() < parallel_batches{

let local_self =self.clone();

let local_batch = batch.to_owned();

let local_args = args.clone();

let local_pipelines = pipelines.clone();

let local_pool = pool.clone();

set.spawn(asyncmove{

local_self

._upsert_documents(local_batch, local_args, local_pipelines, local_pool)

ifletSome(res) = set.join_next().await{

res??;

progress_bar.inc(batch_size);

whileletSome(res) = set.join_next().await{

res??;

progress_bar.inc(batch_size);

progress_bar.println("Done Upserting Documents\n");

progress_bar.finish();

args: serde_json::Map<String,Value>,

) -> anyhow::Result<()>{

let project_info =&self.database_data.as_ref().unwrap().project_info;

let query =if args

.map(|v| v.as_bool().unwrap_or(false))

let batch_size = args

.map(TryToNumeric::try_to_u64)

.unwrap_or(Ok(100))?;

for batchin documents.chunks(batch_sizeasusize){

letmut transaction = pool.begin().await?;

letmut query_values =String::new();

letmut binding_parameter_counter =1;

for _in0..batch.len(){

query_values =format!(

binding_parameter_counter,

binding_parameter_counter +1,

binding_parameter_counter +2

binding_parameter_counter +=3;

letmut transaction = pool.begin().await?;

let query = query.replace(

&query_values.chars().skip(1).collect::<String>(),

let query = query.replace(

letmut query_values =String::new();

letmut binding_parameter_counter =1;

for _in0..batch.len(){

query_values =format!(

binding_parameter_counter,

binding_parameter_counter +1,

binding_parameter_counter +2

binding_parameter_counter +=3;

letmut query = sqlx::query_as(&query);

letmut source_uuids =vec![];

for documentin batch{

let id = document

.context("`id` must be a key in document")?

let md5_digest = md5::compute(id.as_bytes());

let source_uuid = uuid::Uuid::from_slice(&md5_digest.0)?;

source_uuids.push(source_uuid);

let start =SystemTime::now();

let timestamp = start

let versions:HashMap<String, serde_json::Value> = document

.context("document must be an object")?

.try_fold(HashMap::new(), |mut acc,(key, value)|{

let md5_digest = md5::compute(serde_json::to_string(value)?.as_bytes());

let md5_digest =format!("{md5_digest:x}");

acc.insert(

key.to_owned(),

serde_json::json!({

"last_updated": timestamp,

"md5": md5_digest

anyhow::Ok(acc)

})?;

let versions = serde_json::to_value(versions)?;

query = query.bind(source_uuid).bind(document).bind(versions);

let query = query.replace(

&query_values.chars().skip(1).collect::<String>(),

let query = query.replace(

let results:Vec<(i64,Option<Json>)> = query

.bind(source_uuids)

.fetch_all(&mut*transaction)

.await?;

letmut query = sqlx::query_as(&query);

letmut source_uuids =vec![];

for documentin&batch{

let id = document

.context("`id` must be a key in document")?

let md5_digest = md5::compute(id.as_bytes());

let source_uuid = uuid::Uuid::from_slice(&md5_digest.0)?;

source_uuids.push(source_uuid);

let start =SystemTime::now();

let timestamp = start

let versions:HashMap<String, serde_json::Value> = document

.context("document must be an object")?

.try_fold(HashMap::new(), |mut acc,(key, value)|{

let md5_digest = md5::compute(serde_json::to_string(value)?.as_bytes());

let md5_digest =format!("{md5_digest:x}");

acc.insert(

key.to_owned(),

serde_json::json!({

"last_updated": timestamp,

"md5": md5_digest

anyhow::Ok(acc)

})?;

let versions = serde_json::to_value(versions)?;

let dp:Vec<(i64,Json,Option<Json>)> = results

.zip(batch)

.map(|((id, previous_document), document)|{

(id, document.to_owned(), previous_document)

query = query.bind(source_uuid).bind(document).bind(versions);

let results:Vec<(i64,Option<Json>)> = query

.bind(source_uuids)

.fetch_all(&mut*transaction)

.await?;

let dp:Vec<(i64,Json,Option<Json>)> = results

.zip(batch)

.map(|((id, previous_document), document)|(id, document.to_owned(), previous_document))

for(pipeline, parsed_schema)in&mut pipelines{

let ids_to_run_on:Vec<i64> = dp

.filter(|(_, document, previous_document)|match previous_document{

Some(previous_document) => parsed_schema

.any(|(key, _)| document[key] != previous_document[key]),

None =>true,

.map(|(document_id, _, _)|*document_id)

for(pipeline, parsed_schema)in&mut pipelines{

let ids_to_run_on:Vec<i64> = dp

.filter(|(_, document, previous_document)|match previous_document{

Some(previous_document) => parsed_schema

.any(|(key, _)| document[key] != previous_document[key]),

None =>true,

.map(|(document_id, _, _)|*document_id)

if !ids_to_run_on.is_empty(){

pipeline

.sync_documents(ids_to_run_on, project_info,&mut transaction)

if !ids_to_run_on.is_empty(){

pipeline

.sync_documents(ids_to_run_on, project_info,&mut transaction)

transaction.commit().await?;

progress_bar.inc(batch_size);

asyncfncan_add_pipeline_and_upsert_documents_with_parallel_batches() -> anyhow::Result<()>{

let collection_name ="test_r_c_capaud_107";

let pipeline_name ="test_r_p_capaud_6";

letmut pipeline =Pipeline::new(

pipeline_name,

)?;

letmut collection =Collection::new(collection_name,None)?;

collection.add_pipeline(&mut pipeline).await?;

let documents =generate_dummy_documents(20);

collection

documents.clone(),

.await?;

let pool =get_or_initialize_pool(&None).await?;

let documents_table =format!("{}.documents", collection_name);

let queried_documents:Vec<models::Document> =

sqlx::query_as(&query_builder!("SELECT * FROM %s", documents_table))

.fetch_all(&pool)

.await?;

assert!(queried_documents.len() ==20);

let chunks_table =format!("{}_{}.title_chunks", collection_name, pipeline_name);

let title_chunks:Vec<models::Chunk> =

sqlx::query_as(&query_builder!("SELECT * FROM %s", chunks_table))

.fetch_all(&pool)

.await?;

assert!(title_chunks.len() ==20);

let chunks_table =format!("{}_{}.body_chunks", collection_name, pipeline_name);

let body_chunks:Vec<models::Chunk> =

sqlx::query_as(&query_builder!("SELECT * FROM %s", chunks_table))

.fetch_all(&pool)

.await?;

assert!(body_chunks.len() ==120);

let tsvectors_table =format!("{}_{}.body_tsvectors", collection_name, pipeline_name);

let tsvectors:Vec<models::TSVector> =

sqlx::query_as(&query_builder!("SELECT * FROM %s", tsvectors_table))

.fetch_all(&pool)

.await?;

assert!(tsvectors.len() ==120);

collection.archive().await?;

asyncfncan_upsert_documents_and_add_pipeline() -> anyhow::Result<()>{