fix date & add fts index

.github/workflows/docker-build-push.yml

@@ -99,15 +99,15 @@ jobs:
           context: ${{ matrix.context }}
           file: ${{ matrix.dockerfile }}
 
-      - name: delete-untagged-images
-        if: env.GIT_BRANCH_NAME == 'develop'
-        continue-on-error: true
-        uses: actions/delete-package-versions@v5
-        with:
-          package-name: "d2e-trex-base"
-          package-type: "container"
-          min-versions-to-keep: 3
-          delete-only-untagged-versions: "true"
+     # - name: delete-untagged-images
+     #   if: env.GIT_BRANCH_NAME == 'develop'
+     #   continue-on-error: true
+     #   uses: actions/delete-package-versions@v5
+     #   with:
+     #     package-name: "d2e-trex-base"
+     #     package-type: "container"
+     #     min-versions-to-keep: 3
+     #     delete-only-untagged-versions: "true"
 
       - name: npm install
         run: npm install

ext/trex/Cargo.toml

@@ -15,7 +15,8 @@ pgwire = { version = "0.28.0", default-features = false, features = ["server-api
 chrono = {version = "0.4.34", features = ["serde"] }
 tracing-subscriber.workspace = true
 thiserror = "1.0"
-arrow-json = "54.0.0"
+arrow-json = {version = "54.0.0"}
+arrow-array = {version = "54.0.0", features = ["chrono-tz"] }
 async-trait = { workspace = true }
 bigdecimal = { version = "0.4.6", features = ["std"] }
 bytes = { workspace = true }

ext/trex/clients/duckdb.rs

@@ -77,6 +77,26 @@ impl DuckDbClient {
         }
     }
 
+    pub fn create_fts_index(&self, table_name: &TableName) -> Result<bool, duckdb::Error> {
+        if table_name.name.to_lowercase() == "concept" {
+            info!("CREATE FTS");
+            let q0 = "install fts";
+            let c = self.conn.lock().unwrap();
+            let _install_fts = c.execute(q0, []);
+            let query = format!(
+                "PRAGMA create_fts_index({}.{}.{}, {}, {})",
+                &self.current_database, table_name.schema, table_name.name, "concept_id", "'*'"
+            );
+            info!(query);
+            info!("{}", &query);
+            let x = c.execute_batch(&query);
+            info!("RES: {:?}", x);
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
+
     fn postgres_to_duckdb_type(typ: &Type) -> &'static str {
         match typ {
             &Type::BOOL => "bool",

ext/trex/conversions/table.rs

@@ -1,9 +1,10 @@
 use std::fmt::Display;
 
 use pg_escape::quote_identifier;
+use serde::{Deserialize, Serialize};
 use tokio_postgres::types::Type;
 
-#[derive(Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct TableName {
     pub schema: String,
     pub name: String,

ext/trex/js/trex_lib.js

@@ -11,6 +11,7 @@ const {
 	op_prompt,
 	op_prompt_next,
 	op_add_replication,
+	op_copy_tables,
 	op_install_plugin,
 	op_execute_query,
 	op_exit,
@@ -73,10 +74,12 @@ export class DatabaseManager {
 
 	#updatePublications() {
 		for(const c of this.getCredentials()) {
-			if(c.publications) {
-				const adminCredentials = c.credentials.filter(c => c.userScope === 'Admin')[0];
+			const adminCredentials = c.credentials.filter(c => c.userScope === 'Admin')[0];
+
+			if(c.publications && c.publications.length > 0 ) {
+				console.log(`TREX PUB FOUND ${c.id}`)
 				for(const p of c.publications) {
-					const key = `${p.publication}`
+					const key = `${c.id}_${p.publication}`
 					if(!(key in this.getPublications)) {
 						op_add_replication(p.publication, p.slot, key, c.host, c.port, c.name, adminCredentials.username, adminCredentials.password);
 						this.#add_postgres(`${key}_pg`, {host: c.host, port: c.port, databaseName: c.name, user: adminCredentials.username, password: adminCredentials.password});
@@ -85,14 +88,28 @@ export class DatabaseManager {
 						this.#setPublications(pub);
 					}
 				}
-			} 
+			} else {
+				console.log(`TREX NO PUB FOUND ${c.id}`)
+				const key = `${c.id}`
+				if(!(key in this.getPublications)) {
+					this.#add_postgres(`${key}_pg`, {host: c.host, port: c.port, databaseName: c.name, user: adminCredentials.username, password: adminCredentials.password});
+					const res = JSON.parse(op_execute_query(`${key}_pg`,"select table_schema as schema,table_name as name from information_schema.tables where table_type = 'BASE TABLE' and table_schema not in ('information_schema','pg_catalog') ", []));
+					op_copy_tables(res, key, c.host, c.port, c.name, adminCredentials.username, adminCredentials.password);
+					const pub = this.getPublications();
+					pub[key] = true;
+					this.#setPublications(pub);
+				}
+			}
 		}
 	}
 
 	getFirstPublication(db_id) {
-		const tmp =  this.getCredentials().filter(c => c.id === db_id)[0].publications[0]
-		if(tmp)
-			return `${tmp.publication}`
+		try {
+			const tmp =  this.getCredentials().filter(c => c.id === db_id)[0].publications[0]
+			if(tmp)
+				return `${db_id}_${tmp.publication}`
+		} catch(e) {
+		}
 		return `${db_id}`
 	}
 

ext/trex/lib.rs

@@ -4,6 +4,7 @@ pub mod pipeline;
 pub mod sql;
 use std::process;
 
+use conversions::table::TableName;
 use deno_core::error::AnyError;
 use deno_core::op2;
 use duckdb::arrow::record_batch::RecordBatch;
@@ -79,7 +80,9 @@ pub async fn start_sql_server(ip: &str, port: u16, auth_type: AuthType) {
 
 #[derive(Clone)]
 pub enum ReplicateCommand {
-    // CopyTable { schema: String, name: String },
+    CopyTable {
+        tables: Vec<TableName>,
+    },
     Cdc {
         publication: String,
         slot_name: String,
@@ -98,8 +101,8 @@ async fn create_pipeline(
     db_password: Option<String>,
 ) -> Result<BatchDataPipeline<PostgresSource, DuckDbSink>, Box<dyn Error>> {
     let (postgres_source, action) = match command {
-        /* ReplicateCommand::CopyTable { schema, name } => {
-            let table_names = vec![TableName { schema, name }];
+        ReplicateCommand::CopyTable { tables } => {
+            let table_names: Vec<TableName> = tables;
 
             let postgres_source = PostgresSource::new(
                 db_host,
@@ -112,7 +115,7 @@ async fn create_pipeline(
             )
             .await?;
             (postgres_source, PipelineAction::TableCopiesOnly)
-        } */
+        }
         ReplicateCommand::Cdc {
             publication,
             slot_name,
@@ -156,6 +159,9 @@ pub async fn trex_replicate(
 ) -> Result<(), Box<dyn Error>> {
     let mut retries = 0;
     let mut start = SystemTime::now();
+    if matches!(command, ReplicateCommand::CopyTable { tables: _ }) {
+        retries = 4;
+    }
     while retries < 5 {
         let mut pipeline = create_pipeline(
             duckdb,
@@ -170,17 +176,49 @@ pub async fn trex_replicate(
         .await?;
         pipeline.start().await?;
         let duration = SystemTime::now().duration_since(start)?;
-        if duration.as_secs() < 300 {
+        if matches!(command, ReplicateCommand::CopyTable { tables: _ }) {
             retries += 1;
         } else {
-            retries = 0;
-            start = SystemTime::now();
+            if duration.as_secs() < 300 {
+                retries += 1;
+            } else {
+                retries = 0;
+                start = SystemTime::now();
+            }
+            println!("restarting pipeline ... (try {retries})");
         }
-        println!("restarting pipeline ... (try {retries})");
     }
     Ok(())
 }
 
+#[op2]
+fn op_copy_tables(
+    #[serde] tables: Vec<TableName>,
+    #[string] duckdb_file: String,
+    #[string] db_host: String,
+    db_port: u16,
+    #[string] db_name: String,
+    #[string] db_username: String,
+    #[string] db_password: String,
+) {
+    warn!("TREX START TABLE COPY: {duckdb_file}");
+    let command = ReplicateCommand::CopyTable { tables };
+    tokio::spawn(async move {
+        trex_replicate(
+            &TREX_DB,
+            command,
+            duckdb_file.as_str(),
+            db_host.as_str(),
+            db_port,
+            db_name.as_str(),
+            db_username.as_str(),
+            Some(db_password),
+        )
+        .await
+        .map_err(|error| println!("ERROR: {error}"))
+    });
+}
+
 #[allow(clippy::too_many_arguments)]
 #[op2(fast)]
 fn op_add_replication(
@@ -519,7 +557,8 @@ fn op_execute_query(
     let _ = conn
         .execute(&format!("USE {database}"), [])
         .inspect_err(|e| warn!("{e}"));
-    let tmpstmt = conn.prepare(&sql).inspect_err(|e| println!("{e}"));
+
+    let tmpstmt = conn.prepare(&sql).inspect_err(|e| warn!("{e}"));
 
     /*let n = stmt.parameter_count();
     let mut tparams: Vec<TrexType> = Vec::new();
@@ -599,7 +638,8 @@ deno_core::extension!(
         op_execute_query,
         op_exit,
         op_get_dbc,
-        op_set_dbc
+        op_set_dbc,
+        op_copy_tables
     ],
     esm_entry_point = "ext:sb_trex/js/trex_lib.js",
     esm = [

ext/trex/pipeline/sinks/duckdb/executor.rs

@@ -274,6 +274,8 @@ impl DuckDbExecutor {
 
     fn table_copied(&self, table_id: TableId) -> Result<(), DuckDbExecutorError> {
         self.client.insert_into_copied_tables(table_id)?;
+        let table_schema = self.get_table_schema(table_id)?;
+        self.client.create_fts_index(&table_schema.table_name)?;
         Ok(())
     }
 

test/main/test_main.ts

@@ -128,7 +128,7 @@ const init_tests = {
                     }
                 ],
                 "publications": [
-                    {"publication": "test_pub", "slot": "stdout_slot"}
+                    {"publication": "test_pub", "slot": "data2evidence"}
                 ],
                 "extra": [
                     {
@@ -259,6 +259,22 @@ const tests = {
         console.log("Answer:"+res)*/
     },
     "dbquery #6": test_dbquery6,
+    "dbquery json": async () => {
+        const dbm = Trex.userDatabaseManager();
+        const conn = dbm.getConnection('demo_database', 'demo_cdm', "demo_cdm", {"duckdb": (n:any) => n})
+        const res1 = conn.execute("create table if not exists test (id number primary key, test json)",[], ((err:any,res:any) => {
+            console.log(err);
+            if(!err)
+            conn.execute("insert into test values (0, '{\"x\": {\"id\": \"a\"}}')",[], ((err:any,res:any) => {
+                console.log(err);
+                if(!err)
+                conn.execute("select now(), test->'$.x' from test",[], ((err:any,res:any) => {
+                    console.log(res);
+                    console.log(err);
+                      }));
+            }));
+        }));
+    },
 
 }