fix(proxy): use specific EqlTraits on encrypted columns (instead of EqlTrait::all())

Cargo.toml

@@ -43,8 +43,11 @@ debug = true
 
 [workspace.dependencies]
 sqltk = { version = "0.10.0" }
-cipherstash-client = { version = "0.34.0-alpha.4" }
-cts-common = { version = "0.34.0-alpha.4" }
+# TODO: revert to crates.io once the cipherstash-suite version with `IndexType::Ope`
+# is published. See https://github.com/cipherstash/cipherstash-suite for the
+# OPE-enabled change.
+cipherstash-client = { path = "../cipherstash-suite/packages/cipherstash-client" }
+cts-common = { path = "../cipherstash-suite/packages/cts-common" }
 
 thiserror = "2.0.9"
 tokio = { version = "1.44.2", features = ["full"] }

packages/cipherstash-proxy/src/postgresql/context/column.rs

@@ -77,8 +77,8 @@ fn column_type_to_postgres_type(
         (ColumnType::Int, _) => postgres_types::Type::INT4,
         (ColumnType::SmallInt, _) => postgres_types::Type::INT2,
         (ColumnType::Timestamp, _) => postgres_types::Type::TIMESTAMPTZ,
-        (ColumnType::Utf8Str, _) => postgres_types::Type::TEXT,
-        (ColumnType::JsonB, EqlTermVariant::JsonAccessor) => postgres_types::Type::TEXT,
-        (ColumnType::JsonB, _) => postgres_types::Type::JSONB,
+        (ColumnType::Text, _) => postgres_types::Type::TEXT,
+        (ColumnType::Json, EqlTermVariant::JsonAccessor) => postgres_types::Type::TEXT,
+        (ColumnType::Json, _) => postgres_types::Type::JSONB,
     }
 }

packages/cipherstash-proxy/src/postgresql/data/from_sql.rs

@@ -115,7 +115,7 @@ pub fn literal_from_sql(
 ///
 /// | Input Type | Target Column Type | Result |
 /// |------------|--------------------|--------|
-/// | `Type::INT4` | `ColumnType::Utf8Str` | `Plaintext::Utf8Str` |
+/// | `Type::INT4` | `ColumnType::Text` | `Plaintext::Text` |
 /// | `Type::INT2` | `ColumnType::Int` | `Plaintext::Int` |
 /// | `Type::INT8` | `ColumnType::Int` | `Error`` |
 fn text_from_sql(
@@ -126,7 +126,7 @@ fn text_from_sql(
     debug!(target: ENCODING, ?val, ?eql_term, ?col_type);
 
     match (eql_term, col_type) {
-        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Utf8Str) => {
+        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Text) => {
             Ok(Plaintext::new(val))
         }
         (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Float) => {
@@ -168,20 +168,20 @@ fn text_from_sql(
         }
 
         // If JSONB, JSONPATH values are treated as strings
-        (EqlTermVariant::JsonPath | EqlTermVariant::JsonAccessor, ColumnType::JsonB) => {
+        (EqlTermVariant::JsonPath | EqlTermVariant::JsonAccessor, ColumnType::Json) => {
             let val = if val.starts_with("$.") {
                 val.to_string()
             } else {
                 format!("$.{val}")
             };
             Ok(Plaintext::new(val))
         }
-        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::JsonB) => {
+        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Json) => {
             serde_json::from_str::<serde_json::Value>(val)
                 .map_err(|_| MappingError::CouldNotParseParameter)
                 .map(Plaintext::new)
         }
-        (EqlTermVariant::Tokenized, ColumnType::Utf8Str) => Ok(Plaintext::new(val)),
+        (EqlTermVariant::Tokenized, ColumnType::Text) => Ok(Plaintext::new(val)),
 
         (eql_term, col_type) => Err(MappingError::UnsupportedParameterType {
             eql_term,
@@ -202,7 +202,7 @@ fn binary_from_sql(
     debug!(target: ENCODING, ?pg_type, ?eql_term, ?col_type);
 
     match (eql_term, col_type, pg_type) {
-        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Utf8Str, _) => {
+        (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Text, _) => {
             parse_bytes_from_sql::<String>(bytes, pg_type).map(Plaintext::new)
         }
         (EqlTermVariant::Full | EqlTermVariant::Partial, ColumnType::Boolean, _) => {
@@ -253,7 +253,7 @@ fn binary_from_sql(
         }
 
         // If JSONB, JSONPATH values are treated as strings
-        (EqlTermVariant::JsonPath, ColumnType::JsonB, &Type::JSONPATH) => {
+        (EqlTermVariant::JsonPath, ColumnType::Json, &Type::JSONPATH) => {
             parse_bytes_from_sql::<String>(bytes, pg_type).map(|val| {
                 let val = if val.starts_with("$.") {
                     val
@@ -263,7 +263,7 @@ fn binary_from_sql(
                 Plaintext::new(val)
             })
         }
-        (EqlTermVariant::JsonAccessor, ColumnType::JsonB, &Type::TEXT | &Type::VARCHAR) => {
+        (EqlTermVariant::JsonAccessor, ColumnType::Json, &Type::TEXT | &Type::VARCHAR) => {
             parse_bytes_from_sql::<String>(bytes, pg_type).map(|val| {
                 let val = if val.starts_with("$.") {
                     val
@@ -276,7 +276,7 @@ fn binary_from_sql(
         // Python psycopg sends JSON/B as BYTEA
         (
             EqlTermVariant::Full | EqlTermVariant::Partial,
-            ColumnType::JsonB,
+            ColumnType::Json,
             &Type::JSON | &Type::JSONB | &Type::BYTEA,
         ) => parse_bytes_from_sql::<serde_json::Value>(bytes, pg_type).map(Plaintext::new),
 
@@ -356,9 +356,9 @@ fn decimal_from_sql(
             .ok_or(MappingError::CouldNotParseParameter)
             .map(Plaintext::new),
 
-        ColumnType::Utf8Str => Ok(Plaintext::new(decimal.to_string())),
+        ColumnType::Text => Ok(Plaintext::new(decimal.to_string())),
 
-        ColumnType::JsonB => {
+        ColumnType::Json => {
             let val: serde_json::Value = serde_json::from_str(&decimal.to_string())
                 .map_err(|_| MappingError::CouldNotParseParameter)?;
             Ok(Plaintext::new(val))
@@ -408,7 +408,7 @@ mod tests {
             config: ColumnConfig {
                 name: "column".to_owned(),
                 in_place: false,
-                cast_type: ColumnType::Utf8Str,
+                cast_type: ColumnType::Text,
                 indexes: vec![],
                 mode: ColumnMode::PlaintextDuplicate,
             },

packages/cipherstash-proxy/src/postgresql/data/to_sql.rs

@@ -16,7 +16,7 @@ pub fn to_sql(plaintext: &Plaintext, format_code: &FormatCode) -> Result<Option<
 
 fn text_to_sql(plaintext: &Plaintext) -> Result<BytesMut, Error> {
     let s = match &plaintext {
-        Plaintext::Utf8Str(Some(x)) => x.to_string(),
+        Plaintext::Text(Some(x)) => x.to_string(),
         Plaintext::Int(Some(x)) => x.to_string(),
         Plaintext::BigInt(Some(x)) => x.to_string(),
         Plaintext::BigUInt(Some(x)) => x.to_string(),
@@ -26,7 +26,7 @@ fn text_to_sql(plaintext: &Plaintext) -> Result<BytesMut, Error> {
         Plaintext::NaiveDate(Some(x)) => x.to_string(),
         Plaintext::SmallInt(Some(x)) => x.to_string(),
         Plaintext::Timestamp(Some(x)) => x.to_string(),
-        Plaintext::JsonB(Some(x)) => x.to_string(),
+        Plaintext::Json(Some(x)) => x.to_string(),
         _ => "".to_string(),
     };
 
@@ -44,8 +44,8 @@ fn binary_to_sql(plaintext: &Plaintext) -> Result<BytesMut, Error> {
         Plaintext::NaiveDate(x) => x.to_sql_checked(&Type::DATE, &mut bytes),
         Plaintext::SmallInt(x) => x.to_sql_checked(&Type::INT2, &mut bytes),
         Plaintext::Timestamp(x) => x.to_sql_checked(&Type::TIMESTAMPTZ, &mut bytes),
-        Plaintext::Utf8Str(x) => x.to_sql_checked(&Type::TEXT, &mut bytes),
-        Plaintext::JsonB(x) => x.to_sql_checked(&Type::JSONB, &mut bytes),
+        Plaintext::Text(x) => x.to_sql_checked(&Type::TEXT, &mut bytes),
+        Plaintext::Json(x) => x.to_sql_checked(&Type::JSONB, &mut bytes),
         Plaintext::Decimal(x) => x.to_sql_checked(&Type::NUMERIC, &mut bytes),
         // TODO: Implement these
         Plaintext::BigUInt(_x) => unimplemented!(),

packages/cipherstash-proxy/src/proxy/encrypt_config/config.rs

@@ -133,8 +133,8 @@ impl From<CastAs> for ColumnType {
             CastAs::Boolean => ColumnType::Boolean,
             CastAs::Date => ColumnType::Date,
             CastAs::Real | CastAs::Double => ColumnType::Float,
-            CastAs::Text => ColumnType::Utf8Str,
-            CastAs::JsonB => ColumnType::JsonB,
+            CastAs::Text => ColumnType::Text,
+            CastAs::JsonB => ColumnType::Json,
         }
     }
 }
@@ -237,7 +237,7 @@ mod tests {
 
         let column = encrypt_config.get(&ident).expect("column exists");
 
-        assert_eq!(column.cast_type, ColumnType::Utf8Str);
+        assert_eq!(column.cast_type, ColumnType::Text);
         assert!(column.indexes.is_empty());
     }
 

packages/cipherstash-proxy/src/proxy/mod.rs

@@ -63,7 +63,8 @@ impl Proxy {
 
         let encrypt_config_manager = EncryptConfigManager::init(&config.database).await?;
 
-        let schema_manager = SchemaManager::init(&config.database).await?;
+        let schema_manager =
+            SchemaManager::init(&config.database, encrypt_config_manager.clone()).await?;
 
         let eql_version = Proxy::eql_version(&config).await?;
 
@@ -114,8 +115,11 @@ impl Proxy {
                 debug!(msg = "ReloadCommand received", ?command);
                 match command {
                     ReloadCommand::DatabaseSchema(responder) => {
-                        schema_manager.reload().await;
+                        // Reload encrypt config first so the schema reload
+                        // sees the latest per-column index configuration when
+                        // deriving EqlTraits.
                         encrypt_config_manager.reload().await;
+                        schema_manager.reload().await;
                         let _ = responder.send(());
                     }
                     ReloadCommand::EncryptSchema(responder) => {