[BEAM-12400] MongoDBIO support for update within documents #14927

pareshsarafmdb · 2021-06-02T03:12:36Z

Current MongoDBIO connector only supports inserts and replace in a collection. In general there will be requirements for updating an existing document - adding a new field, updating an existing field, pushing into an array or set etc. Added support for this.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Samza	Twister2
Go	---	---	---
Java
Python	---	---	---
XLang		---	---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

…BIO write

pabloem · 2021-06-11T19:28:55Z

can you improve the javadoc as well? It would be helpful if I could see what an example of usage of this transform looksl ike

pareshsarafmdb · 2021-06-15T11:05:50Z

can you improve the javadoc as well? It would be helpful if I could see what an example of usage of this transform looksl ike

Have added javadoc related to update.

pabloem · 2021-06-15T21:05:48Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

+      // builder.add(DisplayData.item("isUpdate", isUpdate()));
+      // builder.add(DisplayData.item("updateKey", updateKey()));
+      // builder.add(DisplayData.item("updateOperator", updateOperator()));
+      // builder.add(Data.item("updateOptions", updateOptions()));
+      // builder.add(DisplayData.item("updateField", updateField()));


Uncoment these? Maybe check if they're not null and uncomment I suppose.

pabloem · 2021-06-15T21:05:59Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

+    public Write withIsUpdate(boolean isUpdate) {
+      return builder().setIsUpdate(isUpdate).build();
+    }
+
+    public Write withUpdateKey(String updateKey) {
+      return builder().setUpdateKey(updateKey).build();
+    }
+
+    public Write withUpdateOperator(String updateOperator) {
+      return builder().setUpdateOperator(updateOperator).build();
+    }
+
+    public Write withUpdateField(String updateField) {
+      return builder().setUpdateField(updateField).build();
+    }
+


Please also add javadoc to each public method

@pareshsarafmdb can you squash your commits? then we'll merge. Thanks!

Hi @pabloem As my changes are on master facing challenges is squashing commits. Will you be able to squash and merge by any chance. If not will try looking for alternatives.

pabloem · 2021-06-15T21:31:56Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

+ * *     .withUpdateKey("key-to-match")
+ * *     .withUpdateField("field-to-update")
+ * *     .withUpdateOperator("$set")
+ * *     .withNumSplits(30))


In this case, only one field can be updated. Is that correct? would it make sense to support updating more than one field?
I see that mongodb works with pairs of updateOperator+updateField. I wonder if it makes sense to support something like that instead?

e.g.:

pipeline .apply(...) .apply(MongoDbIO.write() .withUri("mongodb://localhost:27017") .withDatabase("my-database") .withCollection("my-collection") .withIsUpdate(true) .withUpdateKey("key-to-match") .withUpdateFields( "$set", "field1", "$currentDate", "datefield1", "$min", "special-minimum-field1") .withNumSplits(30))

or perhaps something like this:

.withUpdateFields( UpdateField.of("$set", "field1"), UpdateField.of("$currentDate", "datefield1"), UpdateField.of("$min", "special-minimum-field1")))

I also think it may be helpful to support upsert operations. WDYT?

I don't want to make this too complex - but it feels like these options could be supported easily, right?

Hi Pablo. Thanks for the inputs. I thought the same. But I am confused how the actual values come in? With the single field update whatever value comes in from the pipeline will be set as the value for the field. Now if we have to update for multiple fields how it can be done ? I am not pretty sure how it works in beam. Hope my question is clear.

Upserts functionality can definitely be added.

my guess is that maybe we have an document like this in the database:

{ "name": "ironman", "status": "active", "age": 55, "location": "malibu", "lastUpdated": "(SOME TIMESTAMP)" }

And perhaps the PCollection that comes as input contains the following element:

{ "name": "ironman", "status": "inactive", "age": 56, "location": "malibu" }

In this case, we could do:

.withUpdateKey("name") .withUpdateFields( UpdateField.of("$set", "status"), UpdateField.of("$currentDate", "lastUpdated"), UpdateField.of("$set", "age")))

Does that make sense? Another thing I wonder about is - what if we just want to update all the fields without having to list them all one by one on the updateFields attribute of the transform.

thoughts?

Thanks for the clarification! This makes sense. But in real time Sync (CDC) scenario mostly it will be rows coming in and being added to an array or set within mongoDB. For eg say we have a customer collection. New transactions made by the customer is captured and it should be added to the transactions array within customer document. So we definitely need an option to update with full document. So I think we can add both of these options - updating all the fields together and one by one as well. That will be more feature rich as well. What say ?

and other challenge - right now in your example we are assuming that the field/column name in the source is same as field name to be updated in the target. It can be different in most of the cases. So it might be like below:
.withUpdateKey("name")
.withUpdateFields(
UpdateField.of("$set", "status", "dest-status-field-in-mdb"),
UpdateField.of("$currentDate", "lastUpdated", "dest-lastUpdated-field-in-mdb"),
UpdateField.of("$set", "age", "dest-age-field-in-mdb")))

This makes sense to me! : )

I suppose when we update the full document, then the field names must be the same?

just for my information - is this something that you'll want to take on?

Yeah. Have implemented most of it. Testing it now.

aromanenko-dev

Thanks! I left my comments. Also, please add unit test(s) for this feature.

aromanenko-dev · 2021-06-17T15:59:44Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

@@ -766,6 +800,14 @@ private MongoClient createClient(Read spec) {

      abstract Builder setBatchSize(long batchSize);

+      abstract Builder setIsUpdate(boolean isUpdate);


If all these arguments (isUpdate, updateKey, operator, updateField) should be used only together for update, would it be better to incapsulate it into a functional object and have only one new method, e.g. withUpdate(UpdateConfiguration)?

Thank you for the feedback. This should make it easy and clean. I will work on it.

aromanenko-dev · 2021-06-17T16:01:31Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

        batch.add(new Document(ctx.element()));
        if (batch.size() >= spec.batchSize()) {
-          flush();
+          if (spec.isUpdate()) {


To avoid code copying, please, adjust flush() method accordingly.

@aromanenko-dev @pabloem Have checked in the changes. We have an UpdateConfiguration object where all the update related details are wrapped. Sample below:
MongoDbIO.write().withUri(options.getMongoDBUri())
.withDatabase(options.getDatabase())
.withCollection(options.getCollection()).withUpdateConfiguration(
UpdateConfiguration.create().withUpdateKey("accId").withUpdateFields(
UpdateField.of("$set", "category", "category"),
UpdateField.of("$set","balance", "balance"),
UpdateField.of("$push", "transactions"))
Last update which is a push does update the dest field with full input document. Please review and let me know.

@aromanenko-dev @pabloem gentle reminder to review

@pareshsarafmdb Thanks, let me take a look one more time.
@pabloem Do you have any comments on this?

(Pablo is on vacation until August, please ping again in early August.)

pabloem · 2021-08-09T18:44:33Z

I'll take a look now

pabloem · 2021-08-09T18:44:39Z

Run Java PreCommit

pareshsarafmdb · 2021-08-10T04:23:26Z

Run Java PreCommit

Can you please help me how to do this? Thanks much

aromanenko-dev · 2021-08-10T08:32:00Z

@pareshsarafmdb No worries, it's just a command to run Java PreCommit job. Actually, it fails (you can see the results by clicking on "Details" against Java ("Run Java PreCommit") check):

20:47:11 > Task :sdks:java:io:mongodb:compileJava
20:47:11 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java:43: error: [argument.type.incompatible] incompatible argument for parameter sourceField of UpdateField.
20:47:11     return new UpdateField(updateOperator, null, destField);
20:47:11                                            ^
20:47:11   found   : null (NullType)
20:47:11   required: @Initialized @NonNull String

I'd recommend to run this command ./gradlew :sdks:java:io:mongodb:check locally before pushing your changes to make sure that everything is fine.

pareshsarafmdb · 2021-08-10T10:09:32Z

@pareshsarafmdb No worries, it's just a command to run Java PreCommit job. Actually, it fails (you can see the results by clicking on "Details" against Java ("Run Java PreCommit") check):
20:47:11 > Task :sdks:java:io:mongodb:compileJava
20:47:11 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java:43: error: [argument.type.incompatible] incompatible argument for parameter sourceField of UpdateField.
20:47:11     return new UpdateField(updateOperator, null, destField);
20:47:11                                            ^
20:47:11   found   : null (NullType)
20:47:11   required: @Initialized @NonNull String
I'd recommend to run this command ./gradlew :sdks:java:io:mongodb:check locally before pushing your changes to make sure that everything is fine.

I have done it and verified it again. No issues with the build in local.
BUILD SUCCESSFUL in 3m 4s
58 actionable tasks: 58 executed

pareshsarafmdb · 2021-08-10T10:13:27Z

@pareshsarafmdb No worries, it's just a command to run Java PreCommit job. Actually, it fails (you can see the results by clicking on "Details" against Java ("Run Java PreCommit") check):
20:47:11 > Task :sdks:java:io:mongodb:compileJava
20:47:11 /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Java_Phrase/src/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java:43: error: [argument.type.incompatible] incompatible argument for parameter sourceField of UpdateField.
20:47:11     return new UpdateField(updateOperator, null, destField);
20:47:11                                            ^
20:47:11   found   : null (NullType)
20:47:11   required: @Initialized @NonNull String
I'd recommend to run this command ./gradlew :sdks:java:io:mongodb:check locally before pushing your changes to make sure that everything is fine.
I have done it and verified it again. No issues with the build in local.
BUILD SUCCESSFUL in 3m 4s
58 actionable tasks: 58 executed

And If i see the console log the error doesn't seem to be related to changes I made. Can you pls check. @aromanenko-dev

aromanenko-dev · 2021-08-10T11:40:15Z

Run Java PreCommit

aromanenko-dev · 2021-08-10T16:14:48Z

@pareshsarafmdb Could you rebase against master, run ./gradlew :sdks:java:io:mongodb:compileJava locally and push it if it's ok?

pabloem

Just added minor stylistic comments.

I still wonder: What happens when users want to update the whole old record into the whole new record?

Finally, I have one more question about a later improvement to consider for this IO - is it worth thinking later about passing a lambda to choose the destination collection?

pabloem · 2021-08-10T18:31:26Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java

+  private String updateOperator;
+
+  private String sourceField;
+
+  private String destField;


these can be final, I suppose?

pabloem · 2021-08-10T18:32:28Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java

+
+import java.io.Serializable;
+
+public class UpdateField implements Serializable {


this class could also be an AutoValue class, maybe?

@pabloem Have changed it to auto value class which has two methods - fieldUpdate and fullUpdate. Example below:
MongoDbIO.write()
.withUri("mongodb://localhost:" + port)
.withDatabase(DATABASE)
.withCollection(collectionName)
.withUpdateConfiguration(
UpdateConfiguration.create()
.withUpdateKey("id")
.withUpdateFields(
UpdateField.create().fieldUpdate("$set", "scientist", "scientist"),
UpdateField.create().fieldUpdate("$set", "country", "country"))));
Please review and let me know. @aromanenko-dev Please trigger the build - Error should be hopefully gone now.

pabloem · 2021-08-10T18:40:33Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java

+
+  /** for updating with entire input document. */
+  public static UpdateField of(String updateOperator, String destField) {
+    return new UpdateField(updateOperator, null, destField);


is this used whenever we create a new field? Or whenever the source and dest fields have the same name?

I wonder if it makes sense to add an option for whenever source and destination field have the same name : )

This is used when we need to update full source document into destination document. (Not updating specific fields). Whenever we have to update specific fields we need to specify which field in the source has to be updated to which field in dest document. Do you think there can be a better way?

@pabloem Have changed it to auto value class which has two methods - fieldUpdate and fullUpdate. Example below:
MongoDbIO.write()
.withUri("mongodb://localhost:" + port)
.withDatabase(DATABASE)
.withCollection(collectionName)
.withUpdateConfiguration(
UpdateConfiguration.create()
.withUpdateKey("id")
.withUpdateFields(
UpdateField.create().fieldUpdate("$set", "scientist", "scientist"),
UpdateField.create().fieldUpdate("$set", "country", "country"))));

pabloem · 2021-08-10T18:46:30Z

the precommit seems to be failing due to a static check issue:

pabloem · 2021-08-10T18:46:43Z

... also sorry about the delay! : )

pareshsarafmdb · 2021-08-16T11:12:13Z

Just added minor stylistic comments.

I still wonder: What happens when users want to update the whole old record into the whole new record?

Finally, I have one more question about a later improvement to consider for this IO - is it worth thinking later about passing a lambda to choose the destination collection?

@pabloem If I understand what if user wants to update with entire source document into a destination document - In that case you have a overloaded method which just takes update operator and dest field. This does full document update. Did I make sense?

aaltay · 2021-08-27T00:31:24Z

@pareshsarafmdb - Could you look at the test issues?

pabloem · 2021-08-30T18:54:35Z

test issues are unrelated (at least Java Examples failures). This is on me to review. Sorry about the delay again.

pabloem

Just added a couple comments on the API. Once that's resolveds we can move forward.

pabloem · 2021-08-31T18:06:29Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java

+  public static UpdateField create() {
+    return builder().build();
+  }


This initial constructor creates an UpdateField object that is not valid, right? Doesn't it make sense to make this method private, and then make fullUpdate and fieldUpdate static constructor methods?

Have done the change.

pabloem · 2021-08-31T18:06:54Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateConfiguration.java

+})
+public abstract class UpdateConfiguration implements Serializable {
+
+  abstract @Nullable String updateKey();


can we have more than one field as a key?

Right now only one key which will be matched to _id field in mongodb. We can enhance in further releases is what I thought.

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java

pabloem · 2021-08-31T18:11:05Z

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/UpdateField.java

+  /** Sets the limit of documents to find. */
+  public UpdateField fullUpdate(String updateOperator, String destField) {
+    return toBuilder().setUpdateOperator(updateOperator).setDestField(destField).build();
+  }


In this case, we're inserting the full record into one field of the destination, right?

e.g.:
original

{ "key": "mykey1", "value1": "myvalue1", "value2": ["1", "2"], "value3": "thisvalue" }

newvalue

{ "key": "mykey1", "value1": "myvalue1UPD", "value2": ["1", "2", "3", "4"], "value3": "thisvalueISUPDATED" }

Given configuration UpdateField.fullUpdate("$push", "value4"):
result

{ "key": "mykey1", "value1": "myvalue1", "value2": ["1", "2"], "value3": "thisvalue" "value4": { "key": "mykey1", "value1": "myvalue1UPD", "value2": ["1", "2", "3", "4"], "value3": "thisvalueISUPDATED" } }

Is this correct / intended?

Yes thats right. Thats intended.

pareshsarafmdb · 2021-09-01T05:02:49Z

Just added minor stylistic comments.
I still wonder: What happens when users want to update the whole old record into the whole new record?
Finally, I have one more question about a later improvement to consider for this IO - is it worth thinking later about passing a lambda to choose the destination collection?

@pabloem If I understand what if user wants to update with entire source document into a destination document - In that case you have a overloaded method which just takes update operator and dest field. This does full document update. Did I make sense?

Yes thats right,

pabloem · 2021-09-01T18:01:29Z

okay this LGTM. Happy to merge @aromanenko-dev any other comments? : )

aromanenko-dev · 2021-09-02T12:13:21Z

Thanks, I'm fine with code changes. LGTM.

However, before merging, please, squash all commits and avoid the merge commits - please, use rebase instead.

Also, it's not a good practice to create a PR from master branch, it's much better to use a dedicated feature branch for that. Though, I think we can stand with this for now.

pabloem · 2021-09-02T21:01:38Z

@pareshsarafmdb can you squash your commits? then we'll merge. Thanks!

pabloem · 2021-09-06T03:34:22Z

Run Java PostCommit

pabloem · 2021-09-06T04:57:57Z

squashed using Github's feature

…or update within documents * Did changes for ability to update the existing documents using MongoDBIO write * fixes related to formatting and mongodb update * Updated javadoc for update enhancement * address review comments - MongoDBIO update connector * MongoDBIO - changed UpdateField to be autovalue class * addressed review comments - javadoc, updatefield refactoring

aelmaarouf · 2023-06-17T20:10:44Z

Hello,

I'm sorry to bring up this old discussion but it's the only thing i found so far (i can create an issue if you prefer).
I have a hard time to understand the difference between withFindKey and withUpdateKey can anyone please explain it to me ?
Here is my issue : I have a Mongo document like this

{
 "_id" : objectId,
 "patientId" : "u-u-id",
 "therapyDate" : 1283839993, // timestamp
 "value": x
 }

I know that the combination (patientId and therapyDate) is unique so i am using the following update configuration :

.withFindKey("patientId")
.withFindKey("therapyDate")
.withUpdateFields( UpdateField.fieldUpdate("$set","value", "value"))

Using this configuration a new document is created instead of updating the existing one. Can you point me to the proper direction to update this kind of documents (not depending on the _id as updatekey) ?
thanks in advance

pareshsarafmdb added 2 commits May 21, 2021 12:08

Did changes for ability to update the existing documents using MongoD…

f13eb38

…BIO write

fixes related to formatting and mongodb update

e120a35

aromanenko-dev changed the title ~~MongoDBIO support for update within documents~~ [BEAM-12400] MongoDBIO support for update within documents Jun 4, 2021

aromanenko-dev self-requested a review June 4, 2021 10:24

robertwb requested a review from chamikaramj June 8, 2021 19:55

pabloem self-requested a review June 10, 2021 23:25

Updated javadoc for update enhancement

ccaaf9e

pabloem reviewed Jun 15, 2021

View reviewed changes

pabloem self-requested a review June 15, 2021 21:36

aromanenko-dev requested changes Jun 17, 2021

View reviewed changes

address review comments - MongoDBIO update connector

81ce89b

pabloem reviewed Aug 10, 2021

View reviewed changes

pareshsarafmdb and others added 2 commits August 16, 2021 17:40

MongoDBIO - changed UpdateField to be autovalue class

fdbcb83

Merge branch 'apache:master' into master

1d3d586

pabloem reviewed Aug 31, 2021

View reviewed changes

pareshsarafmdb and others added 2 commits September 1, 2021 11:10

Merge branch 'apache:master' into master

6a2c025

addressed review comments - javadoc, updatefield refactoring

6e7f3bf

pabloem merged commit bcb56b7 into apache:master Sep 6, 2021

		@@ -766,6 +800,14 @@ private MongoClient createClient(Read spec) {

		abstract Builder setBatchSize(long batchSize);

		abstract Builder setIsUpdate(boolean isUpdate);


		import java.io.Serializable;

		public class UpdateField implements Serializable {

[BEAM-12400] MongoDBIO support for update within documents #14927

[BEAM-12400] MongoDBIO support for update within documents #14927

Conversation

pareshsarafmdb commented Jun 2, 2021 • edited by aromanenko-dev

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

pabloem commented Jun 11, 2021

pareshsarafmdb commented Jun 15, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aromanenko-dev left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aromanenko-dev Jul 13, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pabloem commented Aug 9, 2021

pabloem commented Aug 9, 2021

pareshsarafmdb commented Aug 10, 2021

aromanenko-dev commented Aug 10, 2021

pareshsarafmdb commented Aug 10, 2021

pareshsarafmdb commented Aug 10, 2021

aromanenko-dev commented Aug 10, 2021

aromanenko-dev commented Aug 10, 2021

pabloem left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pabloem commented Aug 10, 2021

pabloem commented Aug 10, 2021

pareshsarafmdb commented Aug 16, 2021

aaltay commented Aug 27, 2021

pabloem commented Aug 30, 2021

pabloem left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pareshsarafmdb commented Sep 1, 2021

pabloem commented Sep 1, 2021

aromanenko-dev commented Sep 2, 2021

pabloem commented Sep 2, 2021

pabloem commented Sep 6, 2021

pabloem commented Sep 6, 2021

aelmaarouf commented Jun 17, 2023

pareshsarafmdb commented Jun 2, 2021 •

edited by aromanenko-dev

`ValidatesRunner` compliance status (on master branch)

aromanenko-dev Jul 13, 2021 •

edited