-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for MongoDB as a sink (#181)
- Loading branch information
Showing
15 changed files
with
540 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| --- | ||
| sidebar_position: 2 | ||
| --- | ||
|
|
||
| # MongoDB Connector | ||
|
|
||
| The MongoDB connector allows you to insert data into a MongoDB collection. | ||
| At this time only a sink is implemented, there is no support yet to have MongoDB as a source. | ||
|
|
||
| ## Sink | ||
|
|
||
| The MongoDB sink allows the insertion of data into a MongoDB collection. | ||
|
|
||
| The nuget package for the connector is: | ||
|
|
||
| * FlowtideDotNet.Connector.MongoDB | ||
|
|
||
| Its implementation waits fully until the stream has reached a steady state at a time T until it writes data to the collection. | ||
| This means that its table output can always be traced back to a state from the source systems. | ||
|
|
||
| To use the *MongoDB Sink* add the following line to the *ReadWriteFactory*: | ||
|
|
||
| ```csharp | ||
| factory.AddMongoDbSink("regex pattern for tablename", new FlowtideMongoDBSinkOptions() | ||
| { | ||
| Collection = collection, //MongoDB collection | ||
| Database = databaseName, // MongoDB database | ||
| ConnectionString = connectionString, //Connection string to MongoDB | ||
| PrimaryKeys = primaryKeys //List of columns that will be treated as primary keys in the collection | ||
| }); | ||
| ``` |
47 changes: 47 additions & 0 deletions
47
src/FlowtideDotNet.Connector.MongoDB/Extensions/FlowtideMongoDbReadWriteFactoryExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| // 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. | ||
|
|
||
| using FlowtideDotNet.Connector.MongoDB.Internal; | ||
| using FlowtideDotNet.Core.Engine; | ||
| using FlowtideDotNet.Substrait.Relations; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Text; | ||
| using System.Text.RegularExpressions; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace FlowtideDotNet.Connector.MongoDB.Extensions | ||
| { | ||
| public static class FlowtideMongoDbReadWriteFactoryExtensions | ||
| { | ||
| public static ReadWriteFactory AddMongoDbSink(this ReadWriteFactory factory, string regexPattern, FlowtideMongoDBSinkOptions options, Action<WriteRelation>? transform = null) | ||
| { | ||
| if (regexPattern == "*") | ||
| { | ||
| regexPattern = ".*"; | ||
| } | ||
| factory.AddWriteResolver((writeRel, opt) => | ||
| { | ||
| var regexResult = Regex.Match(writeRel.NamedObject.DotSeperated, regexPattern, RegexOptions.IgnoreCase); | ||
| if (!regexResult.Success) | ||
| { | ||
| return null; | ||
| } | ||
| transform?.Invoke(writeRel); | ||
|
|
||
| return new MongoDBSink(options, writeRel, Core.Operators.Write.ExecutionMode.OnCheckpoint, opt); | ||
| }); | ||
| return factory; | ||
| } | ||
| } | ||
| } |
17 changes: 17 additions & 0 deletions
17
src/FlowtideDotNet.Connector.MongoDB/FlowtideDotNet.Connector.MongoDB.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFramework>net7.0</TargetFramework> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="MongoDB.Driver" Version="2.22.0" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\FlowtideDotNet.Core\FlowtideDotNet.Core.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
31 changes: 31 additions & 0 deletions
31
src/FlowtideDotNet.Connector.MongoDB/FlowtideMongoDBSinkOptions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| // 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. | ||
|
|
||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Text; | ||
| using System.Threading.Tasks; | ||
|
|
||
| namespace FlowtideDotNet.Connector.MongoDB | ||
| { | ||
| public class FlowtideMongoDBSinkOptions | ||
| { | ||
| public string ConnectionString { get; set; } | ||
|
|
||
| public string Database { get; set; } | ||
|
|
||
| public string Collection { get; set; } | ||
|
|
||
| public List<string> PrimaryKeys { get; set; } | ||
| } | ||
| } |
109 changes: 109 additions & 0 deletions
109
src/FlowtideDotNet.Connector.MongoDB/Internal/MongoDBSink.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| // 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. | ||
|
|
||
| using FlowtideDotNet.Core.Operators.Write; | ||
| using FlowtideDotNet.Substrait.Relations; | ||
| using Microsoft.Extensions.Options; | ||
| using MongoDB.Bson; | ||
| using MongoDB.Driver; | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Linq; | ||
| using System.Text; | ||
| using System.Threading.Tasks; | ||
| using System.Threading.Tasks.Dataflow; | ||
|
|
||
| namespace FlowtideDotNet.Connector.MongoDB.Internal | ||
| { | ||
| internal class MongoDBSink : SimpleGroupedWriteOperator | ||
| { | ||
| private readonly FlowtideMongoDBSinkOptions options; | ||
| private readonly WriteRelation writeRelation; | ||
| private readonly StreamEventToBson streamEventToBson; | ||
| private IMongoCollection<BsonDocument> collection; | ||
| List<int> primaryKeys; | ||
|
|
||
| public MongoDBSink(FlowtideMongoDBSinkOptions options, WriteRelation writeRelation, ExecutionMode executionMode, ExecutionDataflowBlockOptions executionDataflowBlockOptions) : base(executionMode, executionDataflowBlockOptions) | ||
| { | ||
| this.options = options; | ||
| this.writeRelation = writeRelation; | ||
| streamEventToBson = new StreamEventToBson(writeRelation.TableSchema.Names); | ||
| } | ||
|
|
||
| public override string DisplayName => "MongoDB Sink"; | ||
|
|
||
| protected override Task<MetadataResult> SetupAndLoadMetadataAsync() | ||
| { | ||
| var connection = new MongoUrl(options.ConnectionString); | ||
| var client = new MongoClient(connection); | ||
| var database = client.GetDatabase(options.Database); | ||
| collection = database.GetCollection<BsonDocument>(options.Collection); | ||
| primaryKeys = new List<int>(); | ||
| foreach (var primaryKey in options.PrimaryKeys) | ||
| { | ||
| var index = writeRelation.TableSchema.Names.FindIndex(x => x.Equals(primaryKey, StringComparison.OrdinalIgnoreCase)); | ||
| if (index < 0) | ||
| { | ||
| throw new InvalidOperationException($"Primary key '{primaryKey}' not found in table schema"); | ||
| } | ||
| primaryKeys.Add(index); | ||
| } | ||
| return Task.FromResult(new MetadataResult(primaryKeys)); | ||
| } | ||
|
|
||
| protected override async Task UploadChanges(IAsyncEnumerable<SimpleChangeEvent> rows) | ||
| { | ||
| List<WriteModel<BsonDocument>> writes = new List<WriteModel<BsonDocument>>(); | ||
| await foreach(var row in rows) | ||
| { | ||
| FilterDefinition<BsonDocument>[] filters = new FilterDefinition<BsonDocument>[primaryKeys.Count]; | ||
| for (int i = 0; i < primaryKeys.Count; i++) | ||
| { | ||
| var pkname = options.PrimaryKeys[i]; | ||
| var col = row.Row.GetColumn(i); | ||
| // Need to take the row value into a bson value | ||
| filters[i] = Builders<BsonDocument>.Filter.Eq(pkname, StreamEventToBson.ToBsonValue(col)); | ||
| } | ||
| FilterDefinition<BsonDocument>? filter = null; | ||
| if (filters.Length > 1) | ||
| { | ||
| filter = Builders<BsonDocument>.Filter.And(filters); | ||
| } | ||
| else | ||
| { | ||
| filter = filters[0]; | ||
| } | ||
| if (row.IsDeleted) | ||
| { | ||
| writes.Add(new DeleteOneModel<BsonDocument>(filter)); | ||
| } | ||
| else | ||
| { | ||
| var doc = streamEventToBson.ToBson(row.Row); | ||
| writes.Add(new ReplaceOneModel<BsonDocument>(filter, doc) { IsUpsert = true }); | ||
| } | ||
|
|
||
| if (writes.Count >= 1000) | ||
| { | ||
| await collection.BulkWriteAsync(writes); | ||
| writes.Clear(); | ||
| } | ||
| } | ||
|
|
||
| if (writes.Count > 0) | ||
| { | ||
| await collection.BulkWriteAsync(writes); | ||
| writes.Clear(); | ||
| } | ||
| } | ||
| } | ||
| } |
91 changes: 91 additions & 0 deletions
91
src/FlowtideDotNet.Connector.MongoDB/Internal/StreamEventToBson.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| // 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. | ||
|
|
||
| using FlexBuffers; | ||
| using FlowtideDotNet.Core; | ||
| using MongoDB.Bson; | ||
|
|
||
| namespace FlowtideDotNet.Connector.MongoDB.Internal | ||
| { | ||
| internal class StreamEventToBson | ||
| { | ||
| private readonly List<string> names; | ||
|
|
||
| public StreamEventToBson(List<string> names) | ||
| { | ||
| this.names = names; | ||
| } | ||
|
|
||
| public BsonDocument ToBson(in StreamEvent streamEvent) | ||
| { | ||
| var doc = new BsonDocument(); | ||
|
|
||
| for (int i = 0; i < names.Count; i++) | ||
| { | ||
| doc.Add(names[i], ToBsonValue(streamEvent.GetColumn(i))); | ||
| } | ||
|
|
||
| return doc; | ||
| } | ||
|
|
||
| public static BsonValue ToBsonValue(FlxValue flxValue) | ||
| { | ||
| switch (flxValue.ValueType) | ||
| { | ||
| case FlexBuffers.Type.Null: | ||
| return BsonNull.Value; | ||
| case FlexBuffers.Type.Bool: | ||
| return new BsonBoolean(flxValue.AsBool); | ||
| case FlexBuffers.Type.Int: | ||
| return new BsonInt64(flxValue.AsLong); | ||
| case FlexBuffers.Type.Uint: | ||
| return new BsonInt64((long)flxValue.AsULong); | ||
| case FlexBuffers.Type.Float: | ||
| return new BsonDouble((long)flxValue.AsULong); | ||
| case FlexBuffers.Type.String: | ||
| return new BsonString(flxValue.AsString); | ||
| case FlexBuffers.Type.Key: | ||
| return new BsonString(flxValue.AsString); | ||
| case FlexBuffers.Type.Blob: | ||
| return new BsonBinaryData(flxValue.AsBlob.ToArray()); | ||
| case FlexBuffers.Type.Decimal: | ||
| return new BsonDecimal128(flxValue.AsDecimal); | ||
| case FlexBuffers.Type.Vector: | ||
| return ToBsonArray(flxValue.AsVector); | ||
| case FlexBuffers.Type.Map: | ||
| return ToBsonDocument(flxValue.AsMap); | ||
| default: | ||
| throw new ArgumentOutOfRangeException(); | ||
| } | ||
| } | ||
|
|
||
| private static BsonValue ToBsonArray(FlxVector values) | ||
| { | ||
| var arr = new BsonArray(); | ||
| for (int i = 0; i < values.Length; i++) | ||
| { | ||
| arr.Add(ToBsonValue(values[i])); | ||
| } | ||
| return arr; | ||
| } | ||
|
|
||
| private static BsonValue ToBsonDocument(FlxMap map) | ||
| { | ||
| BsonDocument doc = new BsonDocument(); | ||
| foreach (var kv in map) | ||
| { | ||
| doc.Add(kv.Key, ToBsonValue(kv.Value)); | ||
| } | ||
| return doc; | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.