ETL step

gistfile1.txt

    public class ProductExtender : AbstractOperation
    {
        private IDocumentStore _store;
        private Type _type;
        private Dictionary<string, string> _fields;

        public ProductExtender(IDocumentStore store, Type type, Dictionary<string, string> fields)
        {
            _store = store;
            _type = type;
            _fields = fields;
        }

        public override IEnumerable<Row> Execute(IEnumerable<Row> rows)
        {
            var newType = _type.FullName + ", " + _type.Assembly.GetName().Name;
            var newEntityName = _store.Conventions.FindTypeTagName(_type);

            foreach (var chunk in rows.Partition(ChunkSize.ItemsStoredPerRavenSave))
            {
                using (var session = _store.OpenSession())
                {
                    List<ICommandData> commands = new List<ICommandData>();

                    foreach (Row record in chunk)
                    {
                        int ProductID = (int)record["i_ProductID"];

                        List<PatchRequest> patchRequests = new List<PatchRequest>(1 + _fields.Count());

                        patchRequests.Add(new PatchRequest()
                            {
                                Type = PatchCommandType.Modify,
                                Name = "@metadata",
                                Value = new RavenJObject(),
                                Nested = new []
                                    {
                                        new PatchRequest()
                                            {
                                                Type = PatchCommandType.Set,
                                                Name = "Raven-Clr-Type",
                                                Value = new RavenJValue(newType),
                                            },
                                        new PatchRequest()
                                            {
                                                Type = PatchCommandType.Set,
                                                Name = "Raven-Entity-Name",
                                                Value = new RavenJValue(newEntityName),
                                            }
                                    }
                            });

                        foreach(var field in _fields)
                            patchRequests.Add(new PatchRequest()
                                {
                                    Type = PatchCommandType.Set,
                                    Name = field.Key,
                                    Value = new RavenJValue(record[field.Value])
                                });

                        commands.Add(new PatchCommandData()
                            {
                                Key = Product.GetIdentity(ProductID),
                                Patches = patchRequests.ToArray()
                            });
                    }

                    session.Advanced.DatabaseCommands.Batch(commands);

                    session.SaveChanges();
                }
            }

            yield break;
        }
    }