Transactional writer support #6
Conversation
|
@iulianov check all this when you get some free time. |
- no synchronization was actually happening. We need a common context for sync to happen.
|
@iulianov we don't have to worry any longer about doc to row and row to doc conversions. It is also done for us. |
|
|
||
| if (withTransaction) { | ||
| df.write | ||
| .format("com.github.anicolaspp.spark.sql.writing.Writer") | ||
| .save(path) | ||
|
|
||
| } else { | ||
| MapRSpark.save(df, path, "_id", false, false) |
There was a problem hiding this comment.
If we want the connector to replace the official connector at some point then the createTable and bulkInsert options should be exposed to the caller. Either through default parameters or through the .option method. Similar to how the current connector allows df.write.option("Operation","Insert").saveToMapRDB("path")
There was a problem hiding this comment.
That is fine. I will create an issue to track that.
GH-9
|
Overall the logic looks good but there are still edge cases that need to be thought about and either addressed through code or documented. You already have one case in the comments but there are others. |
|
|
||
| val store: DocumentStore = connection.getStore(table) | ||
|
|
||
| ids.foreach(store.delete) |
There was a problem hiding this comment.
In a normal rollback this should be fine since the only the confirmed written ids are in the ids value. But if one of the records to be deleted has been removed by an external application then the store.delete should throw an exception. It might need to be caught and ignored here.
One way to prevent a try catch here is to use the checkAndDelete method although that might actually create more overhead.
No description provided.