Distribute values

netfetch/src/ca.rs

@@ -154,6 +154,7 @@ pub async fn ca_search(opts: ListenFromFileOpts) -> Result<(), Error> {
     let d = Database {
         name: "daqbuffer".into(),
         host: "sf-nube-11".into(),
+        port: 5432,
         user: "daqbuffer".into(),
         pass: opts.pg_pass.clone(),
     };
@@ -317,6 +318,7 @@ pub async fn ca_connect(opts: ListenFromFileOpts) -> Result<(), Error> {
     let d = Database {
         name: "daqbuffer".into(),
         host: "sf-nube-11".into(),
+        port: 5432,
         user: "daqbuffer".into(),
         pass: opts.pg_pass.clone(),
     };

netfetch/src/ca/store.rs

@@ -64,7 +64,9 @@ impl DataStore {
             .map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
         let qu_insert_ts_msp = Arc::new(q);
         let q = scy
-            .prepare("insert into series_by_ts_msp (ts_msp, shape_kind, scalar_type, series) values (?, ?, ?, ?)")
+            .prepare(
+                "insert into series_by_ts_msp (part, ts_msp, shape_kind, scalar_type, series) values (?, ?, ?, ?, ?)",
+            )
             .await
             .map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;
         let qu_insert_series_by_ts_msp = Arc::new(q);

netfetch/src/series.rs

@@ -53,8 +53,8 @@ pub async fn get_series_id(pg_client: &PgClient, cd: &ChannelDescDecoded) -> Res
             series &= 0x7fffffffffffffff;
         }
         for _ in 0..2000 {
-            if series > i64::MAX as u64 {
-                series = 0;
+            if series < 1 || series > i64::MAX as u64 {
+                series = 1;
             }
             let res = pg_client
                 .execute(

netfetch/src/store.rs

@@ -196,6 +196,7 @@ pub async fn insert_item(item: InsertItem, data_store: &DataStore, stats: &CaCon
     }
     if let Some(ts_msp_grid) = item.ts_msp_grid {
         let params = (
+            (item.series as i32) & 0xff,
             ts_msp_grid as i32,
             if item.shape.to_scylla_vec().is_empty() { 0 } else { 1 } as i32,
             item.scalar_type.to_scylla_i32(),