Refactor KDB adapter to use q expression strings for writes

- Add to_kdb_row_expr() method to KdbSerialize trait
- Switch from K object construction to q string expressions for inserts
- Use send_sync_message instead of send_async_message for reliability
- Improve dictionary handling in read.rs with cleaner let-chain syntax
- Update integration tests to use new API

Author: 0-jake-0

wingfoil/src/adapters/kdb/read.rs

@@ -102,14 +102,13 @@ where
 fn extract_column_names(result: &K) -> Vec<String> {
     // Tables in KDB (qtype 98) are flipped dictionaries
     // The underlying dictionary has symbol list keys (column names) and list of lists values
-    if result.get_type() == qtype::TABLE {
-        // Get the underlying dictionary from the table
-        if let Ok(dict) = result.get_dictionary() {
-            // Dictionary keys are a K object - extract symbols
-            if let Ok(symbols) = dict.as_vec::<String>() {
-                return symbols.clone();
-            }
-        }
+    if result.get_type() == qtype::TABLE
+        && let Ok(dict) = result.get_dictionary()
+        && let Ok(dict_parts) = dict.as_vec::<K>()
+        && let Some(keys) = dict_parts.first()
+        && let Ok(symbols) = keys.as_vec::<String>()
+    {
+        return symbols.clone();
     }
     Vec::new()
 }
@@ -121,11 +120,11 @@ fn iterate_rows(result: &K) -> Vec<K> {
     // Tables in KDB (qtype 98) are column-oriented
     if result.get_type() == qtype::TABLE
         && let Ok(dict) = result.get_dictionary()
+        && let Ok(dict_parts) = dict.as_vec::<K>()
+        && dict_parts.len() >= 2
     {
-        // The dictionary values are a list of column vectors
-        // We need to access the values side of the dictionary
-        if let Ok(col_list) = dict.as_vec::<K>() {
-            // col_list contains the column vectors
+        let values = &dict_parts[1];
+        if let Ok(col_list) = values.as_vec::<K>() {
             // Get the number of rows from the first column
             if let Some(first_col) = col_list.first() {
                 let n_rows = first_col.len();

wingfoil/src/adapters/kdb/write.rs

@@ -19,6 +19,12 @@ pub trait KdbSerialize: Sized {
     /// The returned K object should be a compound list containing
     /// the column values in the same order as the target table schema.
     fn to_kdb_row(&self) -> K;
+
+    /// Serialize self into a q expression string representing a row.
+    ///
+    /// The returned string should be valid q syntax for a list of values,
+    /// e.g., "(`AAPL; 185.5; 100)" for a trade record.
+    fn to_kdb_row_expr(&self) -> String;
 }
 
 /// Write stream data to a KDB+ table.
@@ -112,18 +118,12 @@ async fn kdb_write_consumer<T>(
     // Process incoming records
     while let Some((_time, batch)) = source.next().await {
         for record in batch {
-            let row = record.to_kdb_row();
-
-            // Build functional query: (insert;`tablename;row)
-            // This is a compound list with: function symbol, table symbol, and row data
-            let query = K::new_compound_list(vec![
-                K::new_symbol("insert".to_string()),
-                K::new_symbol(table_name.clone()),
-                row,
-            ]);
-
-            // Send async message (fire-and-forget insert)
-            if let Err(e) = socket.send_async_message(&query).await {
+            // Build insert query as a string: `tablename insert (values)
+            let row_expr = record.to_kdb_row_expr();
+            let query = format!("`{} insert {}", table_name, row_expr);
+
+            // Send sync message to ensure insert completes before continuing
+            if let Err(e) = socket.send_sync_message(&query.as_str()).await {
                 log::error!("KDB insert failed: {}", e);
             }
         }
@@ -177,6 +177,10 @@ mod tests {
                     K::new_long(self.size),
                 ])
             }
+
+            fn to_kdb_row_expr(&self) -> String {
+                format!("(`{}; {}; {})", self.sym, self.price, self.size)
+            }
         }
 
         let record = TestRecord {
@@ -207,6 +211,10 @@ mod tests {
                     K::new_float(self.price),
                 ])
             }
+
+            fn to_kdb_row_expr(&self) -> String {
+                format!("(`{}; {})", self.sym, self.price)
+            }
         }
 
         let conn = KdbConnection::new("localhost", 5000);

wingfoil/tests/kdb_integration.rs

@@ -52,6 +52,14 @@ impl KdbSerialize for TestTrade {
             K::new_long(self.time),
         ])
     }
+
+    fn to_kdb_row_expr(&self) -> String {
+        // Use {:e} to ensure float format (e.g., 380.0 not 380)
+        format!(
+            "(`{}; {:e}; {}; {})",
+            self.sym, self.price, self.size, self.time
+        )
+    }
 }
 
 impl KdbDeserialize for TestTrade {
@@ -152,15 +160,16 @@ fn test_kdb_round_trip() {
     });
 
     kdb_write(conn.clone(), table_name, &write_stream)
-        .run(
-            RunMode::HistoricalFrom(NanoTime::ZERO),
-            RunFor::Duration(std::time::Duration::from_secs(1)),
-        )
+        .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
         .expect("Write failed");
 
-    // Small delay to ensure writes are flushed
+    // Small delay to ensure async writes are flushed
     std::thread::sleep(std::time::Duration::from_millis(100));
 
+    // Debug: check table contents directly
+    let count: K = rt.block_on(async { run_q(&conn, &format!("count {}", table_name)).await });
+    eprintln!("DEBUG: table count = {:?}", count);
+
     // Read trades back using kdb_read
     let query = format!("select from {}", table_name);
     let read_stream: Rc<dyn Stream<TinyVec<[TestTrade; 1]>>> =
@@ -170,10 +179,7 @@ fn test_kdb_round_trip() {
 
     let collected = read_stream.accumulate();
     collected
-        .run(
-            RunMode::HistoricalFrom(NanoTime::ZERO),
-            RunFor::Duration(std::time::Duration::from_secs(1)),
-        )
+        .run(RunMode::HistoricalFrom(NanoTime::ZERO), RunFor::Forever)
         .expect("Read failed");
 
     // Flatten and compare