fixing tests and migration to the serialized deserialized JSONB2

server/src/tables_data/handlers/put_table_data.rs

@@ -5,6 +5,7 @@ use sqlx::{PgPool, Arguments, Row};
 use sqlx::postgres::PgArguments;
 use chrono::{DateTime, Utc};
 use common::proto::komp_ac::tables_data::{PutTableDataRequest, PutTableDataResponse};
+use common::proto::komp_ac::table_definition::ColumnDefinition;
 
 use std::sync::Arc;
 use prost_types::value::Kind;
@@ -14,6 +15,7 @@ use std::collections::HashMap;
 
 use crate::steel::server::execution::{self, Value};
 use crate::indexer::{IndexCommand, IndexCommandData};
+use crate::table_definition::models::map_field_type;
 use tokio::sync::mpsc;
 use tracing::error;
 
@@ -56,19 +58,20 @@ pub async fn put_table_data(
     .map_err(|e| Status::internal(format!("Table lookup error: {}", e)))?
     .ok_or_else(|| Status::not_found("Table not found"))?;
 
-    // Parse column definitions from JSON format
-    let columns_json: Vec<String> = serde_json::from_value(table_def.columns.clone())
+    // Parse column definitions from JSON format (now ColumnDefinition objects)
+    let stored_columns: Vec<ColumnDefinition> = serde_json::from_value(table_def.columns.clone())
         .map_err(|e| Status::internal(format!("Column parsing error: {}", e)))?;
 
+    // Convert ColumnDefinition → (name, sql_type)
     let mut columns = Vec::new();
-    for col_def in columns_json {
-        let parts: Vec<&str> = col_def.splitn(2, ' ').collect();
-        if parts.len() != 2 {
-            return Err(Status::internal("Invalid column format"));
-        }
-        let name = parts[0].trim_matches('"').to_string();
-        let sql_type = parts[1].to_string();
-        columns.push((name, sql_type));
+    for col_def in stored_columns {
+        let col_name = col_def.name.trim().to_string();
+        let sql_type = map_field_type(&col_def.field_type)
+            .map_err(|e| Status::invalid_argument(format!(
+                "Invalid type for column '{}': {}",
+                col_name, e
+            )))?;
+        columns.push((col_name, sql_type));
     }
 
     // Build list of valid system columns (foreign keys and special columns)