qe: Fix nested objects with $type key in JSON protocol (#4670)

* qe: Fix nested objects with `$type` key in JSON protocol

Introduce another special value to JSON protocol, `"$type": "Raw"`.
When encountered, no other nested `$type` keys would be interpreted as
special and will be written to DB as is. Main usecase is JSON column
values with user-provided `$type` keys.

This is an alternative to #4668, that might look cleaner on the engine
side.

Part of the fix for prisma/prisma#21454, will require client adjustments
as well.

* Fix & move the test

query-engine/connector-test-kit-rs/query-engine-tests/tests/queries/data_types/json.rs

@@ -160,6 +160,62 @@ mod json {
         Ok(())
     }
 
+    #[connector_test]
+    async fn dollar_type_in_json_protocol(runner: Runner) -> TestResult<()> {
+        let res = runner
+            .query_json(
+                r#"{
+                    "modelName": "TestModel",
+                    "action": "createOne",
+                    "query": {
+                       "selection": { "json": true },
+                       "arguments": {
+                          "data": {
+                             "id": 1,
+                             "json": { "$type": "Raw", "value": {"$type": "Something" } }
+                          }
+                       }
+                    }
+                }"#,
+            )
+            .await?;
+
+        res.assert_success();
+
+        insta::assert_snapshot!(res.to_string(), @r###"{"data":{"createOneTestModel":{"json":{"$type":"Json","value":"{\"$type\":\"Something\"}"}}}}"###);
+
+        Ok(())
+    }
+
+    #[connector_test]
+    async fn nested_dollar_type_in_json_protocol(runner: Runner) -> TestResult<()> {
+        let res = runner
+            .query_json(
+                r#"{
+                    "modelName": "TestModel",
+                    "action": "createOne",
+                    "query": {
+                       "selection": { "json": true },
+                       "arguments": {
+                          "data": {
+                             "id": 1,
+                             "json": {
+                                "something": { "$type": "Raw", "value": {"$type": "Something" } }
+                             }
+                          }
+                       }
+                    }
+                }"#,
+            )
+            .await?;
+
+        res.assert_success();
+
+        insta::assert_snapshot!(res.to_string(), @r###"{"data":{"createOneTestModel":{"json":{"$type":"Json","value":"{\"something\":{\"$type\":\"Something\"}}"}}}}"###);
+
+        Ok(())
+    }
+
     async fn create_test_data(runner: &Runner) -> TestResult<()> {
         create_row(runner, r#"{ id: 1, json: "{}" }"#).await?;
         create_row(runner, r#"{ id: 2, json: "{\"a\":\"b\"}" }"#).await?;

query-engine/connector-test-kit-rs/query-tests-setup/src/runner/json_adapter/request.rs

@@ -228,7 +228,7 @@ impl<'a, 'b> FieldTypeInferrer<'a, 'b> {
                     None => InferredType::Unknown,
                 }
             }
-            ArgumentValue::FieldRef(_) => unreachable!(),
+            ArgumentValue::FieldRef(_) | ArgumentValue::Raw(_) => unreachable!(),
         }
     }
 

query-engine/connector-test-kit-rs/query-tests-setup/src/runner/mod.rs

@@ -228,13 +228,14 @@ impl Runner {
         tracing::debug!("Querying: {}", query.clone().green());
 
         println!("{}", query.bright_green());
+        let query: serde_json::Value = serde_json::from_str(&query).unwrap();
 
         let executor = match &self.executor {
             RunnerExecutor::Builtin(e) => e,
-            RunnerExecutor::External(_) => {
+            RunnerExecutor::External(schema_id) => {
                 let response_str: String = executor_process_request(
                     "query",
-                    json!({ "query": query, "txId": self.current_tx_id.as_ref().map(ToString::to_string) }),
+                    json!({ "query": query, "schemaId": schema_id, "txId": self.current_tx_id.as_ref().map(ToString::to_string) }),
                 )
                 .await?;
                 let response: QueryResult = serde_json::from_str(&response_str).unwrap();
@@ -244,7 +245,7 @@ impl Runner {
 
         let handler = RequestHandler::new(&**executor, &self.query_schema, EngineProtocol::Json);
 
-        let serialized_query: JsonSingleQuery = serde_json::from_str(&query).unwrap();
+        let serialized_query: JsonSingleQuery = serde_json::from_value(query).unwrap();
         let request_body = RequestBody::Json(JsonBody::Single(serialized_query));
 
         let result: QueryResult = handler

query-engine/core/src/constants.rs

@@ -11,6 +11,7 @@ pub mod custom_types {
     pub const JSON: &str = "Json";
     pub const ENUM: &str = "Enum";
     pub const FIELD_REF: &str = "FieldRef";
+    pub const RAW: &str = "Raw";
 
     pub fn make_object(typ: &str, value: PrismaValue) -> PrismaValue {
         PrismaValue::Object(vec![make_type_pair(typ), make_value_pair(value)])

query-engine/core/src/query_document/argument_value.rs

@@ -14,6 +14,7 @@ pub enum ArgumentValue {
     Scalar(PrismaValue),
     Object(ArgumentValueObject),
     List(Vec<ArgumentValue>),
+    Raw(serde_json::Value),
     FieldRef(ArgumentValueObject),
 }
 
@@ -46,6 +47,10 @@ impl ArgumentValue {
         Self::Scalar(PrismaValue::Json(str))
     }
 
+    pub fn raw(value: serde_json::Value) -> Self {
+        Self::Raw(value)
+    }
+
     pub fn bytes(bytes: Vec<u8>) -> Self {
         Self::Scalar(PrismaValue::Bytes(bytes))
     }
@@ -76,6 +81,7 @@ impl ArgumentValue {
     pub(crate) fn should_be_parsed_as_json(&self) -> bool {
         match self {
             ArgumentValue::Object(_) => true,
+            ArgumentValue::Raw(_) => true,
             ArgumentValue::List(l) => l.iter().all(|v| v.should_be_parsed_as_json()),
             ArgumentValue::Scalar(pv) => !matches!(pv, PrismaValue::Enum(_) | PrismaValue::Json(_)),
             ArgumentValue::FieldRef(_) => false,

query-engine/core/src/query_document/parser.rs

@@ -871,6 +871,7 @@ pub(crate) mod conversions {
                 format!("({})", itertools::join(v.iter().map(argument_value_to_type_name), ", "))
             }
             ArgumentValue::FieldRef(_) => "FieldRef".to_string(),
+            ArgumentValue::Raw(_) => "JSON".to_string(),
         }
     }
 

query-engine/driver-adapters/executor/src/testd.ts

@@ -69,7 +69,7 @@ async function main(): Promise<void> {
                 debug("[nodejs] Error from request handler: ", err)
                 respondErr(request.id, {
                     code: 1,
-                    message: err.toString(),
+                    message: err.stack ?? err.toString(),
                 })
             }
         } catch (err) {

query-engine/request-handlers/src/protocols/json/protocol_adapter.rs

@@ -199,6 +199,11 @@ impl<'a> JsonProtocolAdapter<'a> {
                         .map(ArgumentValue::float)
                         .map_err(|_| build_err())
                 }
+
+                Some(custom_types::RAW) => {
+                    let value = obj.get(custom_types::VALUE).ok_or_else(build_err)?;
+                    Ok(ArgumentValue::raw(value.clone()))
+                }
                 Some(custom_types::BYTES) => {
                     let value = obj
                         .get(custom_types::VALUE)