mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 04:10:49 +00:00
fix(mongodb): support array filters in updates
This commit is contained in:
@@ -13,6 +13,7 @@ import com.mongodb.MongoClientSettings;
|
||||
import com.mongodb.ServerAddress;
|
||||
import com.mongodb.client.MongoClient;
|
||||
import com.mongodb.client.MongoClients;
|
||||
import com.mongodb.client.model.UpdateOptions;
|
||||
import java.io.BufferedReader;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.InputStream;
|
||||
@@ -625,15 +626,47 @@ public final class MongoAgent {
|
||||
String filterJson = params.get("filter_json").getAsString();
|
||||
String updateJson = params.get("update_json").getAsString();
|
||||
boolean many = params.get("many").getAsBoolean();
|
||||
String optionsJson = params.has("options_json") && !params.get("options_json").isJsonNull()
|
||||
? params.get("options_json").getAsString()
|
||||
: null;
|
||||
|
||||
var col = c.getDatabase(database).getCollection(collection);
|
||||
Document filter = documentForWrite(filterJson);
|
||||
Document update = documentForWrite(updateJson);
|
||||
requireBulkUpdateOperatorDocument(update);
|
||||
var result = many ? col.updateMany(filter, update) : col.updateOne(filter, update);
|
||||
UpdateOptions options = updateOptionsForWrite(optionsJson);
|
||||
var result = many ? col.updateMany(filter, update, options) : col.updateOne(filter, update, options);
|
||||
return Collections.singletonMap("modified_count", result.getModifiedCount());
|
||||
}
|
||||
|
||||
static UpdateOptions updateOptionsForWrite(String optionsJson) {
|
||||
UpdateOptions result = new UpdateOptions();
|
||||
if (optionsJson == null || optionsJson.trim().isEmpty()) {
|
||||
return result;
|
||||
}
|
||||
Document options = Document.parse(optionsJson);
|
||||
for (String key : options.keySet()) {
|
||||
if (!"arrayFilters".equals(key)) {
|
||||
throw new IllegalArgumentException("Unsupported update option: " + key);
|
||||
}
|
||||
}
|
||||
Object rawFilters = options.get("arrayFilters");
|
||||
if (rawFilters == null) {
|
||||
return result;
|
||||
}
|
||||
if (!(rawFilters instanceof List<?>)) {
|
||||
throw new IllegalArgumentException("arrayFilters must be an array");
|
||||
}
|
||||
List<Document> filters = new ArrayList<>();
|
||||
for (Object filter : (List<?>) rawFilters) {
|
||||
if (!(filter instanceof Document)) {
|
||||
throw new IllegalArgumentException("Each arrayFilters entry must be an object");
|
||||
}
|
||||
filters.add((Document) filter);
|
||||
}
|
||||
return result.arrayFilters(filters);
|
||||
}
|
||||
|
||||
static Document documentForWrite(String docJson) {
|
||||
Document doc = Document.parse(docJson);
|
||||
convertMongoShellDates(doc);
|
||||
|
||||
@@ -12,6 +12,7 @@ import com.google.gson.JsonArray;
|
||||
import com.google.gson.JsonObject;
|
||||
import com.google.gson.JsonParser;
|
||||
import com.mongodb.MongoClientSettings;
|
||||
import com.mongodb.client.model.UpdateOptions;
|
||||
import java.io.FileInputStream;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
@@ -202,6 +203,25 @@ class MongoAgentTest {
|
||||
assertFalse(json.getAsJsonObject("error").get("message").getAsString().contains("Unknown method"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesArrayFiltersUpdateOption() {
|
||||
UpdateOptions options = MongoAgent.updateOptionsForWrite(
|
||||
"{\"arrayFilters\":[{\"item.id\":322678}]}"
|
||||
);
|
||||
|
||||
assertEquals(1, options.getArrayFilters().size());
|
||||
assertEquals(322678, ((Document) options.getArrayFilters().get(0)).getInteger("item.id"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void rejectsUnsupportedUpdateOptions() {
|
||||
IllegalArgumentException error = assertThrows(
|
||||
IllegalArgumentException.class,
|
||||
() -> MongoAgent.updateOptionsForWrite("{\"upsert\":true}")
|
||||
);
|
||||
assertEquals("Unsupported update option: upsert", error.getMessage());
|
||||
}
|
||||
|
||||
@Test
|
||||
void deleteDocumentsMethodIsRecognizedOverJsonRpc() {
|
||||
String response = MongoAgent.handleRequest(
|
||||
|
||||
@@ -2098,8 +2098,8 @@ export async function documentUpdateDocument(connectionId: string, database: str
|
||||
return post("/api/document-store/update-document", { connectionId, database, collection, id, docJson, routing });
|
||||
}
|
||||
|
||||
export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean): Promise<{ affected_rows: number }> {
|
||||
return post("/api/mongo/update-documents", { connectionId, database, collection, filterJson, updateJson, many });
|
||||
export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean, optionsJson?: string): Promise<{ affected_rows: number }> {
|
||||
return post("/api/mongo/update-documents", { connectionId, database, collection, filterJson, updateJson, many, optionsJson });
|
||||
}
|
||||
|
||||
export async function mongoDeleteDocument(connectionId: string, database: string, collection: string, id: string, routing?: string): Promise<number> {
|
||||
|
||||
@@ -1869,7 +1869,7 @@ export async function documentUpdateDocument(connectionId: string, database: str
|
||||
return invoke("document_update_document", { connectionId, database, collection, id, docJson, routing });
|
||||
}
|
||||
|
||||
export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean): Promise<{ affected_rows: number }> {
|
||||
export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean, optionsJson?: string): Promise<{ affected_rows: number }> {
|
||||
const affectedRows = await invoke<number>("mongo_update_documents", {
|
||||
connectionId,
|
||||
database,
|
||||
@@ -1877,6 +1877,7 @@ export async function mongoUpdateDocuments(connectionId: string, database: strin
|
||||
filterJson,
|
||||
updateJson,
|
||||
many,
|
||||
optionsJson,
|
||||
});
|
||||
return { affected_rows: affectedRows };
|
||||
}
|
||||
|
||||
@@ -50,7 +50,7 @@ export type MongoCommand =
|
||||
| ({ kind: "collectionStats" } & MongoCollectionStatsCommand)
|
||||
| ({ kind: "use" } & MongoUseCommand)
|
||||
| { kind: "insert"; collection: string; docsJson: string }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; many: boolean }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; options?: string; many: boolean }
|
||||
| { kind: "delete"; collection: string; filter: string; many: boolean }
|
||||
| { kind: "createIndex"; collection: string; keys: string; options?: string }
|
||||
| { kind: "dropIndex"; collection: string; index: string }
|
||||
@@ -293,11 +293,13 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null
|
||||
const target = parseCollectionMethodTarget(source, method);
|
||||
if (!target) continue;
|
||||
const args = parseMethodArgs(source, target.methodCallIndex);
|
||||
if (!args || args.length !== 2) return null;
|
||||
if (!args || args.length < 2 || args.length > 3) return null;
|
||||
const filter = normalizeJsonArgument(args[0]);
|
||||
const update = normalizeJsonArgument(args[1]);
|
||||
if (!filter || !update) return null;
|
||||
return { kind: "update", collection: target.collection, filter, update, many: method === "updateMany" };
|
||||
const options = args[2]?.trim() ? normalizeJsonArgument(args[2]) : undefined;
|
||||
if (args[2]?.trim() && !options) return null;
|
||||
return { kind: "update", collection: target.collection, filter, update, ...(options ? { options } : {}), many: method === "updateMany" };
|
||||
}
|
||||
|
||||
for (const method of ["deleteOne", "deleteMany"] as const) {
|
||||
|
||||
@@ -2732,7 +2732,7 @@ export const useQueryStore = defineStore("query", () => {
|
||||
const result = await api.mongoInsertDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.docsJson);
|
||||
allResults.push(markQueryResultRowsRaw(annotateMongoResult(mongoWriteToQueryResult(result.affected_rows, performance.now() - commandStartedAt))));
|
||||
} else if (mongoCommand.kind === "update") {
|
||||
const result = await api.mongoUpdateDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.filter, mongoCommand.update, mongoCommand.many);
|
||||
const result = await api.mongoUpdateDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.filter, mongoCommand.update, mongoCommand.many, mongoCommand.options);
|
||||
allResults.push(markQueryResultRowsRaw(annotateMongoResult(mongoWriteToQueryResult(result.affected_rows, performance.now() - commandStartedAt))));
|
||||
} else if (mongoCommand.kind === "createIndex") {
|
||||
const result = await api.mongoCreateIndex(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.keys, mongoCommand.options);
|
||||
|
||||
@@ -905,6 +905,7 @@ pub async fn update_documents(
|
||||
filter_json: &str,
|
||||
update_json: &str,
|
||||
many: bool,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<u64, String> {
|
||||
let filter_value: serde_json::Value =
|
||||
serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
@@ -912,15 +913,47 @@ pub async fn update_documents(
|
||||
serde_json::from_str(update_json).map_err(|e| format!("Invalid update JSON: {e}"))?;
|
||||
let filter = json_filter_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
|
||||
let update = json_object_to_document(&update_value).map_err(|e| format!("Invalid update: {e}"))?;
|
||||
let array_filters = parse_update_array_filters(options_json)?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result = if many {
|
||||
col.update_many(filter, update).await.map_err(|e| e.to_string())?
|
||||
let mut action = col.update_many(filter, update);
|
||||
if let Some(filters) = array_filters {
|
||||
action = action.array_filters(filters);
|
||||
}
|
||||
action.await.map_err(|e| e.to_string())?
|
||||
} else {
|
||||
col.update_one(filter, update).await.map_err(|e| e.to_string())?
|
||||
let mut action = col.update_one(filter, update);
|
||||
if let Some(filters) = array_filters {
|
||||
action = action.array_filters(filters);
|
||||
}
|
||||
action.await.map_err(|e| e.to_string())?
|
||||
};
|
||||
Ok(result.modified_count)
|
||||
}
|
||||
|
||||
#[derive(Default, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct MongoUpdateOptions {
|
||||
array_filters: Option<Vec<serde_json::Value>>,
|
||||
}
|
||||
|
||||
fn parse_update_array_filters(options_json: Option<&str>) -> Result<Option<Vec<Document>>, String> {
|
||||
let Some(raw) = options_json.filter(|value| !value.trim().is_empty()) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let options: MongoUpdateOptions = serde_json::from_str(raw).map_err(|e| format!("Invalid update options: {e}"))?;
|
||||
options
|
||||
.array_filters
|
||||
.map(|filters| {
|
||||
filters
|
||||
.iter()
|
||||
.map(json_filter_to_document)
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| format!("Invalid arrayFilters: {e}"))
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
pub async fn delete_document(client: &Client, database: &str, collection: &str, id: &str) -> Result<u64, String> {
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
for filter in document_id_filters(id) {
|
||||
@@ -1248,6 +1281,23 @@ fn expand_object_id_string_array(items: &[serde_json::Value]) -> Bson {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn update_options_parse_array_filters() {
|
||||
let filters = parse_update_array_filters(Some(r#"{"arrayFilters":[{"item.id":322678},{"item.active":true}]}"#))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(filters, vec![doc! { "item.id": 322678_i64 }, doc! { "item.active": true }]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn update_options_reject_unsupported_fields() {
|
||||
let error = parse_update_array_filters(Some(r#"{"upsert":true}"#)).unwrap_err();
|
||||
|
||||
assert!(error.starts_with("Invalid update options:"));
|
||||
assert!(error.contains("unknown field `upsert`"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multi_seed_uri_removes_direct_connection_true_before_driver_parse() {
|
||||
let uri =
|
||||
|
||||
@@ -315,12 +315,14 @@ pub async fn mongo_update_documents_core(
|
||||
filter_json: &str,
|
||||
update_json: &str,
|
||||
many: bool,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<u64, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::update_documents(client, database, collection, filter_json, update_json, many).await
|
||||
mongo_driver::update_documents(client, database, collection, filter_json, update_json, many, options_json)
|
||||
.await
|
||||
}
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
@@ -331,6 +333,7 @@ pub async fn mongo_update_documents_core(
|
||||
"filter_json": filter_json,
|
||||
"update_json": update_json,
|
||||
"many": many,
|
||||
"options_json": options_json,
|
||||
}))
|
||||
.await?;
|
||||
Ok(result.get("modified_count").and_then(|v| v.as_u64()).unwrap_or(0))
|
||||
|
||||
@@ -154,6 +154,7 @@ pub struct MongoUpdateDocumentsRequest {
|
||||
pub filter_json: String,
|
||||
pub update_json: String,
|
||||
pub many: bool,
|
||||
pub options_json: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -439,6 +440,7 @@ pub async fn update_documents(
|
||||
&req.filter_json,
|
||||
&req.update_json,
|
||||
req.many,
|
||||
req.options_json.as_deref(),
|
||||
)
|
||||
.await
|
||||
.map_err(AppError)?;
|
||||
|
||||
+21
@@ -143,6 +143,27 @@ test("parseMongoWriteCommand accepts unquoted insert and update commands", () =>
|
||||
});
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand accepts updateMany arrayFilters options", () => {
|
||||
assert.deepEqual(
|
||||
parseMongoWriteCommand(`db.issue_3231.updateMany(
|
||||
{ msgType: 3, "order.orderId": { $in: [12345] } },
|
||||
{ $set: { "order.$[orderElem].bcorderproducts.$[prodElem].pankouType": "双双2" } },
|
||||
{ arrayFilters: [
|
||||
{ "orderElem.orderId": { $in: [12345] } },
|
||||
{ "prodElem.id": 322678 }
|
||||
] }
|
||||
)`),
|
||||
{
|
||||
kind: "update",
|
||||
collection: "issue_3231",
|
||||
filter: '{ "msgType": 3, "order.orderId": { "$in": [12345] } }',
|
||||
update: '{ "$set": { "order.$[orderElem].bcorderproducts.$[prodElem].pankouType": "双双2" } }',
|
||||
options: '{ "arrayFilters": [\n { "orderElem.orderId": { "$in": [12345] } },\n { "prodElem.id": 322678 }\n ] }',
|
||||
many: true,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand parses createIndex with optional options", () => {
|
||||
assert.deepEqual(parseMongoWriteCommand("db.users.createIndex({email: 1}, {unique: true, name: 'users_email_unique'})"), {
|
||||
kind: "createIndex",
|
||||
|
||||
@@ -1163,6 +1163,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
filter_json: command.filter,
|
||||
update_json: command.update,
|
||||
many: command.many,
|
||||
options_json: command.options,
|
||||
});
|
||||
return { affectedRows: result.affected_rows };
|
||||
}
|
||||
@@ -1316,7 +1317,7 @@ interface MongoCollectionStatsCommand {
|
||||
|
||||
export type MongoWriteCommand =
|
||||
| { kind: "insert"; collection: string; docsJson: string }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; many: boolean }
|
||||
| { kind: "update"; collection: string; filter: string; update: string; options?: string; many: boolean }
|
||||
| { kind: "delete"; collection: string; filter: string; many: boolean }
|
||||
| { kind: "createIndex"; collection: string; keys: string; options?: string }
|
||||
| { kind: "dropIndex"; collection: string; index: string }
|
||||
@@ -1465,11 +1466,13 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null
|
||||
const target = parseCollectionMethodTarget(source, method);
|
||||
if (!target) continue;
|
||||
const args = parseMethodArgs(source, target.methodCallIndex);
|
||||
if (!args || args.length !== 2) return null;
|
||||
if (!args || args.length < 2 || args.length > 3) return null;
|
||||
const filter = normalizeJsonArgument(args[0]);
|
||||
const update = normalizeJsonArgument(args[1]);
|
||||
if (!filter || !update) return null;
|
||||
return { kind: "update", collection: target.collection, filter, update, many: method === "updateMany" };
|
||||
const options = args[2]?.trim() ? normalizeJsonArgument(args[2]) : undefined;
|
||||
if (args[2]?.trim() && !options) return null;
|
||||
return { kind: "update", collection: target.collection, filter, update, ...(options ? { options } : {}), many: method === "updateMany" };
|
||||
}
|
||||
|
||||
for (const method of ["deleteOne", "deleteMany"] as const) {
|
||||
|
||||
@@ -375,6 +375,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo
|
||||
filterJson: command.filter,
|
||||
updateJson: command.update,
|
||||
many: command.many,
|
||||
optionsJson: command.options,
|
||||
}),
|
||||
});
|
||||
const result = (await res.json()) as { affected_rows: number };
|
||||
|
||||
+16
@@ -73,6 +73,22 @@ test("parseMongoWriteCommand accepts unquoted update operator keys", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("parseMongoWriteCommand accepts updateMany arrayFilters options", () => {
|
||||
assert.deepEqual(
|
||||
parseMongoWriteCommand(
|
||||
'db.orders.updateMany({status: "open"}, {$set: {"items.$[item].status": "done"}}, {arrayFilters: [{"item.id": 7}]})',
|
||||
),
|
||||
{
|
||||
kind: "update",
|
||||
collection: "orders",
|
||||
filter: '{"status": "open"}',
|
||||
update: '{"$set": {"items.$[item].status": "done"}}',
|
||||
options: '{"arrayFilters": [{"item.id": 7}]}',
|
||||
many: true,
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
test("parseMongoCountDocumentsCommand accepts shell-style count commands", () => {
|
||||
assert.deepEqual(parseMongoCountDocumentsCommand('db.projects.countDocuments({"active":true})'), {
|
||||
collection: "projects",
|
||||
|
||||
@@ -134,6 +134,7 @@ struct MongoUpdateDocumentsRequest {
|
||||
filter_json: String,
|
||||
update_json: String,
|
||||
many: bool,
|
||||
options_json: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -722,6 +723,7 @@ async fn handle_mongo_update_documents_data(state: &Arc<AppState>, body: &str, s
|
||||
&req.filter_json,
|
||||
&req.update_json,
|
||||
req.many,
|
||||
req.options_json.as_deref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
|
||||
@@ -263,6 +263,7 @@ pub async fn mongo_update_documents(
|
||||
filter_json: String,
|
||||
update_json: String,
|
||||
many: bool,
|
||||
options_json: Option<String>,
|
||||
) -> Result<u64, String> {
|
||||
ensure_connection_writable(&state, &connection_id, "Update").await?;
|
||||
dbx_core::mongo_ops::mongo_update_documents_core(
|
||||
@@ -273,6 +274,7 @@ pub async fn mongo_update_documents(
|
||||
&filter_json,
|
||||
&update_json,
|
||||
many,
|
||||
options_json.as_deref(),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user