updateMany reworked (legacy+newstyle)

This commit is contained in:
Stefano Lottini committed 2026-09-28 10:30:22 +02:00
1 parent 61c517e21a
commit 7117dfaf63
4 files changed
+85 -11

No files matched your search

@@ -56,6 +56,7 @@ public class DataApiOpMapper implements OpMapper<DataApiBaseOp,DataApiSpace> {
DEPRECATED_OP_TYPES.put(DataApiOpType.find_one_and_update, DataApiOpType.collection_find_one_and_update);
DEPRECATED_OP_TYPES.put(DataApiOpType.replace_one, null); // not a Data API primitive, it's client-emulated
DEPRECATED_OP_TYPES.put(DataApiOpType.find_one_and_replace, DataApiOpType.collection_find_one_and_replace);
DEPRECATED_OP_TYPES.put(DataApiOpType.update_many, DataApiOpType.collection_update_many);
DEPRECATED_OP_TYPES.put(DataApiOpType.count_documents, DataApiOpType.collection_count_documents);
DEPRECATED_OP_TYPES.put(DataApiOpType.estimated_document_count, DataApiOpType.collection_estimated_document_count);
}
@@ -108,7 +109,6 @@ public class DataApiOpMapper implements OpMapper<DataApiBaseOp,DataApiSpace> {
case insert_many -> new DataApiCollectionInsertManyOpDispenser(adapter, op, typeAndTarget.targetFunction);
case insert_one -> new DataApiCollectionInsertOneOpDispenser(adapter, op, typeAndTarget.targetFunction);
case insert_one_vector -> new DataApiCollectionInsertOneVectorOpDispenser(adapter, op, typeAndTarget.targetFunction);
case update_many -> new DataApiCollectionUpdateManyOpDispenser(adapter, op, typeAndTarget.targetFunction);
// LEGACY OPS
// admin ops:
@@ -130,6 +130,7 @@ public class DataApiOpMapper implements OpMapper<DataApiBaseOp,DataApiSpace> {
case find_one_and_update -> new DataApiCollectionLegacyFindOneAndUpdateOpDispenser(adapter, op, typeAndTarget.targetFunction);
case replace_one -> new DataApiCollectionLegacyReplaceOneOpDispenser(adapter, op, typeAndTarget.targetFunction);
case find_one_and_replace -> new DataApiCollectionLegacyFindOneAndReplaceOpDispenser(adapter, op, typeAndTarget.targetFunction);
case update_many -> new DataApiCollectionLegacyUpdateManyOpDispenser(adapter, op, typeAndTarget.targetFunction);
case count_documents -> new DataApiCollectionLegacyCountDocumentsOpDispenser(adapter, op, typeAndTarget.targetFunction);
case estimated_document_count ->
new DataApiCollectionEstimatedDocumentCountOpDispenser(adapter, op, typeAndTarget.targetFunction);
@@ -144,6 +145,7 @@ public class DataApiOpMapper implements OpMapper<DataApiBaseOp,DataApiSpace> {
case collection_delete_one -> new DataApiCollectionDeleteOneOpDispenser(adapter, op, typeAndTarget.targetFunction);
case collection_find_one_and_delete -> new DataApiCollectionFindOneAndDeleteOpDispenser(adapter, op, typeAndTarget.targetFunction);
case collection_find_one_and_replace -> new DataApiCollectionFindOneAndReplaceOpDispenser(adapter, op, typeAndTarget.targetFunction);
case collection_update_many -> new DataApiCollectionUpdateManyOpDispenser(adapter, op, typeAndTarget.targetFunction);
case collection_estimated_document_count ->
new DataApiCollectionEstimatedDocumentCountOpDispenser(adapter, op, typeAndTarget.targetFunction);
case collection_count_documents -> new DataApiCollectionCountDocumentsOpDispenser(adapter, op, typeAndTarget.targetFunction);
@@ -0,0 +1,74 @@
/*
* Copyright (c) nosqlbench
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.nosqlbench.adapter.dataapi.opdispensers;
import com.datastax.astra.client.databases.Database;
import com.datastax.astra.client.core.query.Filter;
import com.datastax.astra.client.collections.commands.Update;
import com.datastax.astra.client.collections.commands.options.CollectionUpdateManyOptions;
import io.nosqlbench.adapter.dataapi.DataApiDriverAdapter;
import io.nosqlbench.adapter.dataapi.ops.DataApiBaseOp;
import io.nosqlbench.adapter.dataapi.ops.DataApiCollectionUpdateManyOp;
import io.nosqlbench.adapters.api.templating.ParsedOp;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.Map;
import java.util.Optional;
import java.util.function.LongFunction;
public class DataApiCollectionLegacyUpdateManyOpDispenser extends DataApiOpDispenser {
private static final Logger logger = LogManager.getLogger(DataApiCollectionLegacyUpdateManyOpDispenser.class);
private final LongFunction<DataApiCollectionUpdateManyOp> opFunction;
public DataApiCollectionLegacyUpdateManyOpDispenser(DataApiDriverAdapter adapter, ParsedOp op, LongFunction<String> targetFunction) {
super(adapter, op, targetFunction);
this.opFunction = createOpFunction(op);
}
private LongFunction<DataApiCollectionUpdateManyOp> createOpFunction(ParsedOp op) {
return (l) -> {
Database db = spaceFunction.apply(l).getDatabase();
Filter filter = getLegacyFilterFromOp(op, l);
CollectionUpdateManyOptions options = getCollectionUpdateManyOptions(op, l);
LongFunction<Map> docMapFunc = op.getAsRequiredFunction("updates", Map.class);
return new DataApiCollectionUpdateManyOp(
db,
db.getCollection(targetFunction.apply(l)),
filter,
new Update(docMapFunc.apply(l)),
options
);
};
}
private CollectionUpdateManyOptions getCollectionUpdateManyOptions(ParsedOp op, long l) {
CollectionUpdateManyOptions options = new CollectionUpdateManyOptions();
Optional<LongFunction<Boolean>> upsertFunction = op.getAsOptionalFunction("upsert", Boolean.class);
if (upsertFunction.isPresent()) {
options = options.upsert(upsertFunction.get().apply(l));
}
return options;
}
@Override
public DataApiBaseOp getOp(long cycle) {
return opFunction.apply(cycle);
}
}
@@ -27,8 +27,6 @@ import io.nosqlbench.adapters.api.templating.ParsedOp;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.Map;
import java.util.Optional;
import java.util.function.LongFunction;
public class DataApiCollectionUpdateManyOpDispenser extends DataApiOpDispenser {
@@ -45,13 +43,13 @@ public class DataApiCollectionUpdateManyOpDispenser extends DataApiOpDispenser {
Database db = spaceFunction.apply(l).getDatabase();
Filter filter = getFilterFromOp(op, l);
CollectionUpdateManyOptions options = getCollectionUpdateManyOptions(op, l);
LongFunction<Map> docMapFunc = op.getAsRequiredFunction("updates", Map.class);
Update update = getUpdateFromOp(op, l);
return new DataApiCollectionUpdateManyOp(
db,
db.getCollection(targetFunction.apply(l)),
filter,
new Update(docMapFunc.apply(l)),
update,
options
);
};
@@ -59,11 +57,10 @@ public class DataApiCollectionUpdateManyOpDispenser extends DataApiOpDispenser {
private CollectionUpdateManyOptions getCollectionUpdateManyOptions(ParsedOp op, long l) {
CollectionUpdateManyOptions options = new CollectionUpdateManyOptions();
Optional<LongFunction<Boolean>> upsertFunction = op.getAsOptionalFunction("upsert", Boolean.class);
if (upsertFunction.isPresent()) {
options = options.upsert(upsertFunction.get().apply(l));
Boolean upsert = getUpsertFromOp(op, l);
if (upsert != null) {
options = options.upsert(upsert);
}
return options;
}
@@ -37,8 +37,6 @@ public enum DataApiOpType {
insert_many,
insert_one,
insert_one_vector,
update_many,
replace_one,
// LEGACY OPS
// admin ops:
@@ -58,7 +56,9 @@ public enum DataApiOpType {
find_one,
update_one,
find_one_and_update,
replace_one,
find_one_and_replace,
update_many,
estimated_document_count,
count_documents,
@@ -75,6 +75,7 @@ public enum DataApiOpType {
collection_update_one,
collection_find_one_and_update,
collection_find_one_and_replace,
collection_update_many,
collection_estimated_document_count,
collection_count_documents,
}