Support row-level DELETEs in MongoDB connector

[NaveenNitturu]

When executing a select statement with an 'OR' condition, it is working well.

trino> select * from mongo.admin.sales where emp_id=1 or num=102;
 emp_id |  email   | num 
--------+----------+-----
      1 | abc@mail | 101 
      2 | def@mail | 102 
(2 rows)

However, when attempting to execute a delete statement with the same 'OR' condition, an error is encountered.

trino> delete from mongo.admin.sales where emp_id=1 or num=102;
Query 20240209_100530_00002_rsh84 failed: This connector does not support modifying table rows
io.trino.spi.TrinoException: This connector does not support modifying table rows
        at io.trino.spi.connector.ConnectorMetadata.beginMerge(ConnectorMetadata.java:929)
        at io.trino.tracing.TracingConnectorMetadata.beginMerge(TracingConnectorMetadata.java:769)
        at io.trino.metadata.MetadataManager.beginMerge(MetadataManager.java:1278)
        at io.trino.tracing.TracingMetadata.beginMerge(TracingMetadata.java:802)
        at io.trino.sql.planner.optimizations.BeginTableWrite$Rewriter.createWriterTarget(BeginTableWrite.java:261)
        at io.trino.sql.planner.optimizations.BeginTableWrite$Rewriter.visitTableFinish(BeginTableWrite.java:188)
        at io.trino.sql.planner.optimizations.BeginTableWrite$Rewriter.visitTableFinish(BeginTableWrite.java:106)
        at io.trino.sql.planner.plan.TableFinishNode.accept(TableFinishNode.java:105)
        at io.trino.sql.planner.plan.SimplePlanRewriter$RewriteContext.rewrite(SimplePlanRewriter.java:81)
        at io.trino.sql.planner.plan.SimplePlanRewriter$RewriteContext.lambda$defaultRewrite$0(SimplePlanRewriter.java:72)
        at com.google.common.collect.ImmutableList.forEach(ImmutableList.java:422)
        at io.trino.sql.planner.plan.SimplePlanRewriter$RewriteContext.defaultRewrite(SimplePlanRewriter.java:72)
        at io.trino.sql.planner.plan.SimplePlanRewriter.visitPlan(SimplePlanRewriter.java:37)
        at io.trino.sql.planner.plan.SimplePlanRewriter.visitPlan(SimplePlanRewriter.java:21)
        at io.trino.sql.planner.plan.PlanVisitor.visitOutput(PlanVisitor.java:49)
        at io.trino.sql.planner.plan.OutputNode.accept(OutputNode.java:82)
        at io.trino.sql.planner.plan.SimplePlanRewriter.rewriteWith(SimplePlanRewriter.java:31)
        at io.trino.sql.planner.optimizations.BeginTableWrite.optimize(BeginTableWrite.java:91)
        at io.trino.sql.planner.LogicalPlanner.runOptimizer(LogicalPlanner.java:309)
        at io.trino.sql.planner.LogicalPlanner.plan(LogicalPlanner.java:270)
        at io.trino.sql.planner.LogicalPlanner.plan(LogicalPlanner.java:239)
        at io.trino.sql.planner.LogicalPlanner.plan(LogicalPlanner.java:234)
        at io.trino.execution.SqlQueryExecution.doPlanQuery(SqlQueryExecution.java:486)
        at io.trino.execution.SqlQueryExecution.planQuery(SqlQueryExecution.java:466)
        at io.trino.execution.SqlQueryExecution.start(SqlQueryExecution.java:404)
        at io.trino.execution.SqlQueryManager.createQuery(SqlQueryManager.java:264)
        at io.trino.dispatcher.LocalDispatchQuery.startExecution(LocalDispatchQuery.java:145)
        at io.trino.dispatcher.LocalDispatchQuery.lambda$waitForMinimumWorkers$2(LocalDispatchQuery.java:129)
        at io.airlift.concurrent.MoreFutures.lambda$addSuccessCallback$12(MoreFutures.java:568)
        at io.airlift.concurrent.MoreFutures$3.onSuccess(MoreFutures.java:543)
        at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1133)
        at io.trino.$gen.Trino_dev____20240209_094727_2.run(Unknown Source)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:842)
        Suppressed: java.lang.Exception: Current plan:
                Output[columnNames = [rows]]
                │   Layout: [rows:bigint]
                └─ TableCommit[target = mongo:MongoTableHandle{schemaTableName=admin.sales, remoteTableName=admin.sales, filter=Optional.empty, constraint=ALL, projectedColumns=[], limit=OptionalInt.empty}]
                   │   Layout: [rows:bigint]
                   └─ LocalExchange[partitioning = SINGLE]
                      │   Layout: [partialrows:bigint, fragment:varbinary]
                      └─ RemoteExchange[type = GATHER]
                         │   Layout: [partialrows:bigint, fragment:varbinary]
                         └─ MergeWriter[table = mongo:MongoTableHandle{schemaTableName=admin.sales, remoteTableName=admin.sales, filter=Optional.empty, constraint=ALL, projectedColumns=[], limit=OptionalInt.empty}]
                            │   Layout: [partialrows:bigint, fragment:varbinary]
                            └─ LocalExchange[partitioning = SINGLE]
                               │   Layout: [emp_id:integer, email:varchar, operation:tinyint, field:bigint, insert_from_update:tinyint]
                               └─ RemoteExchange[partitionCount = 100, type = REPARTITION]
                                  │   Layout: [emp_id:integer, email:varchar, operation:tinyint, field:bigint, insert_from_update:tinyint]
                                  └─ ScanFilterProject[table = mongo:MongoTableHandle{schemaTableName=admin.sales, remoteTableName=admin.sales, filter=Optional.empty, constraint=ALL, projectedColumns=[emp_id:integer, num:integer, $merge_row_id:bigint], limit=OptionalInt.empty}, filterPredicate = (("emp_id" = 1) OR ("num" = 102))]
                                         Layout: [emp_id:integer, email:varchar, operation:tinyint, field:bigint, insert_from_update:tinyint]
                                         emp_id := CAST(null AS integer)
                                         email := CAST(null AS varchar)
                                         operation := TINYINT '2'
                                         insert_from_update := TINYINT '0'
                                         field := $merge_row_id:bigint
                                         num := num:integer
                                         emp_id := emp_id:integer

                at io.trino.sql.planner.optimizations.BeginTableWrite.optimize(BeginTableWrite.java:97)
                ... 17 more

The regular delete operation is functioning properly, and the OR condition with the same columns is also working correctly.

trino> delete from mongo.admin.sales where emp_id=1;
DELETE: 1 row

trino> delete from mongo.admin.sales where emp_id=2 or emp_id=3;
DELETE: 2 rows

Could anyone help on this?

[hashhar]

It seems you're using the MongoDB connector.

It doesn't support DELETE when the filter cannot be pushed down.

In delete from mongo.admin.sales where emp_id=2 or emp_id=3; the filter can be simplified to emp_id IN (2, 3) which is something we can push down to MongoDB.

In the failing query you have OR on two different columns which isn't something we can push down to MongoDB yet.

I'm changing the title to reflect what the actual ask is here but no one is actively working on this at the moment.

[noah-instructure]

In the Trino documentation, this is specified for other connectors but not for MongoDB

[abusk]

@hashhar i want to work on it, can you please suggest good starting point?

[feealexandre]

Executing the following query

DELETE FROM mongo.table1
WHERE _id IN (
  SELECT t2._id
  FROM mongo.table2 t2
  LIMIT 10
)

This connector does not support modifying table rows

[hashhar]

@feealexandre this was already explained in the older comment that if the predicate (in this case an entire subquery) cannot be pushed down to the source then DELETE won't work with MongoDB connector. This matches your example.

For people who are interested in working on this see #23034 for example of what it takes to add it to a JDBC based connector.

[chenjian2664]

Hi, I have tried on #23770. Welcome for the reviews