Add KDB+ write support

- Add NanoTime::to_kdb_timestamp() for converting to KDB timestamps
- Add KdbSerialize trait for serializing Rust types to KDB rows
- Add kdb_write() function to write stream data to KDB+ tables
- Add KdbWriteOperators trait for fluent API (.kdb_write())
- Add filter_map operator to StreamOperators

Author: 0-jake-0

wingfoil/src/adapters/kdb.rs

@@ -1,4 +1,4 @@
-//! KDB+ database adapter for reading data from q/kdb+ instances.
+//! KDB+ database adapter for reading and writing data to q/kdb+ instances.
 //!
 //! This module provides async connectivity to KDB+ databases using the `kdbplus` crate,
 //! allowing query results to be streamed into wingfoil graphs.
@@ -36,11 +36,13 @@
 //!     .unwrap();
 //! ```
 
-use crate::nodes::produce_async;
+use crate::nodes::{FutStream, StreamOperators, produce_async};
 use crate::types::*;
+use futures::StreamExt;
 use kdbplus::ipc::{ConnectionMethod, K, QStream};
 use kdbplus::qtype;
 use std::fmt;
+use std::pin::Pin;
 use std::rc::Rc;
 use tinyvec::TinyVec;
 
@@ -210,6 +212,151 @@ where
     })
 }
 
+/// Trait for serializing Rust types to KDB row data.
+///
+/// Implementors should create a K object representing a single row
+/// that can be inserted into a KDB table.
+pub trait KdbSerialize: Sized {
+    /// Serialize self into a K object representing a row.
+    ///
+    /// 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;
+}
+
+/// Write stream data to a KDB+ table.
+///
+/// This function connects to a KDB+ instance and inserts records from the
+/// upstream stream into the specified table. Each batch of records is
+/// inserted using the `` `tablename insert row`` syntax.
+///
+/// # Type Parameters
+/// * `T` - The record type, must implement `Element`, `Send`, and `KdbSerialize`
+///
+/// # Arguments
+/// * `connection` - KDB connection configuration
+/// * `table_name` - Name of the target table
+/// * `upstream` - Stream of record batches to insert
+///
+/// # Returns
+/// A Node that drives the write operation.
+///
+/// # Example
+///
+/// ```ignore
+/// use wingfoil::adapters::kdb::*;
+/// use wingfoil::*;
+///
+/// #[derive(Debug, Clone, Default)]
+/// struct Trade {
+///     sym: String,
+///     price: f64,
+///     size: i64,
+/// }
+///
+/// impl KdbSerialize for Trade {
+///     fn to_kdb_row(&self) -> K {
+///         K::new_compound_list(vec![
+///             K::new_symbol(self.sym.clone()),
+///             K::new_float(self.price),
+///             K::new_long(self.size),
+///         ])
+///     }
+/// }
+///
+/// let conn = KdbConnection::new("localhost", 5000);
+/// let trades: Rc<dyn Stream<TinyVec<[Trade; 1]>>> = /* ... */;
+/// kdb_write(conn, "trades", &trades)
+///     .run(RunMode::RealTime, RunFor::Duration(Duration::from_secs(10)))
+///     .unwrap();
+/// ```
+pub fn kdb_write<T>(
+    connection: KdbConnection,
+    table_name: impl Into<String>,
+    upstream: &Rc<dyn Stream<TinyVec<[T; 1]>>>,
+) -> Rc<dyn Node>
+where
+    T: Element + Send + KdbSerialize + 'static,
+{
+    let table_name = table_name.into();
+
+    let consumer = Box::new(move |source: Pin<Box<dyn FutStream<TinyVec<[T; 1]>>>>| {
+        kdb_write_consumer(connection, table_name, source)
+    });
+
+    upstream.consume_async(consumer)
+}
+
+async fn kdb_write_consumer<T>(
+    connection: KdbConnection,
+    table_name: String,
+    mut source: Pin<Box<dyn FutStream<TinyVec<[T; 1]>>>>,
+) where
+    T: Element + Send + KdbSerialize + 'static,
+{
+    let creds = connection.credentials_string();
+
+    // Connect to KDB
+    let mut socket = match QStream::connect(
+        ConnectionMethod::TCP,
+        &connection.host,
+        connection.port,
+        &creds,
+    )
+    .await
+    {
+        Ok(s) => s,
+        Err(e) => {
+            log::error!("KDB connection failed: {}", e);
+            return;
+        }
+    };
+
+    // 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 {
+                log::error!("KDB insert failed: {}", e);
+            }
+        }
+    }
+}
+
+/// Extension trait for writing streams to KDB+ tables.
+///
+/// This trait provides a fluent API for writing `TinyVec<[T; 1]>` streams
+/// (output of `combine`, `kdb_read`, etc.) to KDB+ tables.
+pub trait KdbWriteOperators<T: Element> {
+    /// Write this stream to a KDB+ table.
+    ///
+    /// # Arguments
+    /// * `conn` - KDB connection configuration
+    /// * `table` - Name of the target table
+    ///
+    /// # Returns
+    /// A Node that drives the write operation.
+    fn kdb_write(self: &Rc<Self>, conn: KdbConnection, table: &str) -> Rc<dyn Node>;
+}
+
+impl<T: Element + Send + KdbSerialize + 'static> KdbWriteOperators<T>
+    for dyn Stream<TinyVec<[T; 1]>>
+{
+    fn kdb_write(self: &Rc<Self>, conn: KdbConnection, table: &str) -> Rc<dyn Node> {
+        kdb_write(conn, table, self)
+    }
+}
+
 /// Extract column names from a KDB table result.
 ///
 /// For tables, the result is a flipped dictionary where the keys are column names.
@@ -376,4 +523,88 @@ mod tests {
         let nano = NanoTime::from_kdb_timestamp(kdb_time);
         assert_eq!(u64::from(nano), 946_684_801_000_000_000);
     }
+
+    #[test]
+    fn test_nanotime_kdb_timestamp_round_trip() {
+        // Test round-trip conversion: NanoTime -> KDB timestamp -> NanoTime
+        let original = NanoTime::new(1_000_000_000_000_000_000); // Some Unix timestamp
+        let kdb_ts = original.to_kdb_timestamp();
+        let restored = NanoTime::from_kdb_timestamp(kdb_ts);
+        assert_eq!(original, restored);
+
+        // Test with KDB epoch (2000-01-01)
+        let kdb_epoch = NanoTime::new(946_684_800_000_000_000);
+        assert_eq!(kdb_epoch.to_kdb_timestamp(), 0);
+
+        // Test with values after KDB epoch
+        let after_epoch = NanoTime::new(946_684_801_000_000_000); // 1 second after
+        assert_eq!(after_epoch.to_kdb_timestamp(), 1_000_000_000);
+    }
+
+    #[test]
+    fn test_kdb_serialize_trait() {
+        // Test that KdbSerialize trait can be implemented
+        #[derive(Debug, Clone, Default)]
+        struct TestRecord {
+            sym: String,
+            price: f64,
+            size: i64,
+        }
+
+        impl KdbSerialize for TestRecord {
+            fn to_kdb_row(&self) -> K {
+                K::new_compound_list(vec![
+                    K::new_symbol(self.sym.clone()),
+                    K::new_float(self.price),
+                    K::new_long(self.size),
+                ])
+            }
+        }
+
+        let record = TestRecord {
+            sym: "AAPL".to_string(),
+            price: 185.50,
+            size: 100,
+        };
+
+        let row = record.to_kdb_row();
+        assert_eq!(row.get_type(), qtype::COMPOUND_LIST);
+    }
+
+    #[test]
+    fn test_kdb_write_node_creation() {
+        use crate::nodes::constant;
+
+        // Test that kdb_write creates a valid node
+        #[derive(Debug, Clone, Default)]
+        struct TestTrade {
+            sym: String,
+            price: f64,
+        }
+
+        impl KdbSerialize for TestTrade {
+            fn to_kdb_row(&self) -> K {
+                K::new_compound_list(vec![
+                    K::new_symbol(self.sym.clone()),
+                    K::new_float(self.price),
+                ])
+            }
+        }
+
+        let conn = KdbConnection::new("localhost", 5000);
+
+        // Create a constant stream that produces TinyVec batches
+        let trade = TestTrade {
+            sym: "TEST".to_string(),
+            price: 100.0,
+        };
+        let mut batch: TinyVec<[TestTrade; 1]> = TinyVec::new();
+        batch.push(trade);
+
+        let stream = constant(batch);
+
+        // Verify we can create the kdb_write node (doesn't require actual KDB connection)
+        let _node = kdb_write(conn, "test_table", &stream);
+        // Node creation succeeds - actual connection would be attempted during run()
+    }
 }

wingfoil/src/nodes/mod.rs

@@ -323,6 +323,12 @@ pub trait StreamOperators<T: Element> {
     /// Map's it's source into a new Stream using the supplied closure.
     fn map<OUT: Element>(self: &Rc<Self>, func: impl Fn(T) -> OUT + 'static)
     -> Rc<dyn Stream<OUT>>;
+    /// Combines map and filter: transforms and optionally filters values.
+    /// The closure returns (OUT, bool) where bool indicates whether to emit.
+    fn filter_map<OUT: Element>(
+        self: &Rc<Self>,
+        func: impl Fn(T) -> (OUT, bool) + 'static,
+    ) -> Rc<dyn Stream<OUT>>;
     /// Map's source into a new Stream using a fallible closure.
     /// Errors propagate to graph execution.
     fn try_map<OUT: Element>(
@@ -532,6 +538,13 @@ where
         MapStream::new(self.clone(), Box::new(func)).into_stream()
     }
 
+    fn filter_map<OUT: Element>(
+        self: &Rc<Self>,
+        func: impl Fn(T) -> (OUT, bool) + 'static,
+    ) -> Rc<dyn Stream<OUT>> {
+        MapFilterStream::new(self.clone(), Box::new(func)).into_stream()
+    }
+
     fn try_map<OUT: Element>(
         self: &Rc<Self>,
         func: impl Fn(T) -> anyhow::Result<OUT> + 'static,

wingfoil/src/time.rs

@@ -54,6 +54,11 @@ impl NanoTime {
     pub fn from_kdb_timestamp(kdb_nanos: i64) -> Self {
         Self::new((kdb_nanos as u64).wrapping_add(Self::KDB_EPOCH_OFFSET_NANOS))
     }
+
+    /// Convert to KDB timestamp (nanoseconds from 2000-01-01).
+    pub fn to_kdb_timestamp(self) -> i64 {
+        (self.0.wrapping_sub(Self::KDB_EPOCH_OFFSET_NANOS)) as i64
+    }
 }
 
 impl From<u128> for NanoTime {