public SqlEntityDelete(AbstractConnection connection, Entity entity)
     : base(connection) {
     _name = Connection.Enclose(entity.OutputName());
     _isMaster = entity.IsMaster();
     BatchSize = connection.BatchSize;
     UseTransaction = true;
 }

        public override void WriteEndVersion(Process process, AbstractConnection input, Entity entity, bool force = false) {
            if (entity.Updates + entity.Inserts > 0 || force) {
                var client = new ElasticSearchClientFactory().Create(this, TflBatchEntity(entity.ProcessName));
                var versionType = entity.Version == null ? "string" : entity.Version.SimpleType;

                string end;
                if (versionType.Equals("datetime") && entity.End is DateTime) {
                    end = ((DateTime)entity.End).ToString("yyyy-MM-ddTHH:mm:ss.fff");
                } else if (versionType.Equals("byte[]") || versionType.Equals("rowversion")) {
                    end = Common.BytesToHexString((byte[])entity.End);
                } else {
                    end = new DefaultFactory(Logger).Convert(entity.End, versionType).ToString();
                }

                var body = new {
                    id = entity.TflBatchId,
                    tflbatchid = entity.TflBatchId,
                    process = entity.ProcessName,
                    connection = input.Name,
                    entity = entity.Alias,
                    updates = entity.Updates,
                    inserts = entity.Inserts,
                    deletes = entity.Deletes,
                    version = end,
                    version_type = versionType,
                    tflupdate = DateTime.UtcNow
                };
                client.Client.Index(client.Index, client.Type, body);
            }
        }

 public SqlServerEntityBatchUpdate(AbstractConnection connection, Entity entity)
     : base(connection) {
     _connection = connection;
     _entity = entity;
     BatchSize = connection.BatchSize;
     UseTransaction = true;
 }

        public Fields Read(AbstractConnection connection, string process, string prefix, string name, string schema, bool isMaster = false) {
            var fields = new Fields();

            using (var cn = connection.GetConnection()) {
                cn.Open();
                var sql = PrepareSql();
                connection.Logger.EntityDebug(name, sql);

                var results = cn.Query(sql, new { name, schema });

                foreach (var result in results) {
                    var columnName = result.COLUMN_NAME;
                    var type = GetSystemType(result.DATA_TYPE);
                    var length = result.CHARACTER_MAXIMUM_LENGTH == "0" || result.CHARACTER_MAXIMUM_LENGTH == "-1" ? "64" : result.CHARACTER_MAXIMUM_LENGTH;
                    var fieldType = (bool)result.IS_PRIMARY_KEY ? (isMaster ? FieldType.MasterKey : FieldType.PrimaryKey) : FieldType.NonKey;
                    var field = new Field(type, length, fieldType, true, string.Empty) {
                        Name = columnName,
                        Entity = name,
                        Process = process,
                        Index = Convert.ToInt16(result.ORDINAL_POSITION - 1),
                        Schema = schema,
                        Input = true,
                        Precision = result.NUMERIC_PRECISION,
                        Scale = result.NUMERIC_SCALE,
                        Alias = prefix + columnName
                    };
                    fields.Add(field);
                }
            }

            return fields;
        }

 public static FSDirectory Create(AbstractConnection connection) {
     var directoryInfo = new DirectoryInfo(connection.Folder);
     if (!directoryInfo.Exists) {
         directoryInfo.Create();
     }
     return FSDirectory.Open(directoryInfo);
 }

 public bool Exists(AbstractConnection connection, Entity entity) {
     if (!new DirectoryInfo(LuceneDirectoryFactory.Path(connection, entity)).Exists)
         return false;
     using (var dir = LuceneDirectoryFactory.Create(connection, entity)) {
         return dir.ListAll().Length > 0;
     }
 }

 public VistaSystemFileHandler(AbstractConnection cxn)
 {
     myCxn = cxn;
     getFileDefs();
     files = new Hashtable();
     lookupTables = new Hashtable();
 }

 public SqlOverrideOperation(Entity entity, AbstractConnection connection)
     : base(connection) {
     CommandBehavior = CommandBehavior.Default;
     EntityName = entity.Name;
     _entity = entity;
     _fields = entity.Fields.WithInput().NameAliases().ToArray();
 }

 public bool Exists(AbstractConnection connection, Entity entity) {
     using (var cn = connection.GetConnection()) {
         cn.Open();
         var table = cn.Query<string>(SQL, new { name = entity.OutputName() }).DefaultIfEmpty(string.Empty).FirstOrDefault();
         return !string.IsNullOrEmpty(table);
     }
 }

        public IScriptReponse Execute(AbstractConnection connection, string script, int timeOut = 0) {
            var response = new ScriptResponse();
            var server = new Server();
            var logger = connection.Logger;

            try {
                logger.Debug("Connecting to {0} on {1}.", connection.Database, connection.Server);
                server.Connect(connection.GetConnectionString());

                var results = server.Execute(script);

                foreach (XmlaResult result in results) {
                    foreach (XmlaMessage message in result.Messages) {
                        response.Messages.Add(message.Description);
                    }
                }
                response.Success = response.Messages.Count == 0;
            } catch (Exception e) {
                logger.Debug(e.Message + (e.InnerException != null ? " " + e.InnerException.Message : string.Empty));
                response.Messages.Add(e.Message);
            } finally {
                if (server.Connected) {
                    logger.Debug("Disconnecting from {0} on {1}.", connection.Database, connection.Server);
                    server.Disconnect();
                }
            }
            return response;
        }

        public override void WriteEndVersion(Process process, AbstractConnection input, Entity entity, bool force = false) {

            if (entity.Updates + entity.Inserts > 0 || force) {

                var solr = GetOperations(process, entity.OutputName());
                var versionType = entity.Version == null ? "string" : entity.Version.SimpleType;

                string end;
                if (versionType.Equals("datetime") && entity.End is DateTime) {
                    end = ((DateTime)entity.End).ToString("yyyy-MM-ddTHH:mm:ss.fff");
                } else if (versionType.Equals("byte[]") || versionType.Equals("rowversion")) {
                    end = Common.BytesToHexString((byte[])entity.End);
                } else {
                    end = new DefaultFactory(Logger).Convert(entity.End, versionType).ToString();
                }

                var doc = new Dictionary<string, object> {
                    { "id", entity.TflBatchId},
                    { "tflbatchid", entity.TflBatchId},
                    { "process", entity.ProcessName},
                    { "connection", input.Name},
                    { "entity", entity.Alias},
                    { "updates", entity.Updates},
                    { "inserts", entity.Inserts},
                    { "deletes", entity.Deletes},
                    { "version", end},
                    { "version_type", versionType},
                    { "tflupdate", DateTime.UtcNow}
                };
                solr.Add(doc);
                solr.Commit();
            }
        }

 public bool Check(AbstractConnection connection) {
     var result = connection.Name.Equals("output", IC) || new DirectoryInfo(connection.Folder).Exists;
     if (!result) {
         _logger.Warn("Could not verify that '{0}' exists.", connection.Folder);
     }
     return result;
 }

        public override void WriteEndVersion(Process process, AbstractConnection input, Entity entity, bool force = false) {
            if (entity.Updates + entity.Inserts <= 0 && !force)
                return;

            var versionType = entity.Version == null ? "string" : entity.Version.SimpleType;
            var end = entity.End ?? new DefaultFactory(Logger).Convert(entity.End, versionType);

            using (var dir = LuceneDirectoryFactory.Create(this, TflBatchEntity(entity.ProcessName))) {
                using (var writer = new IndexWriter(dir, new KeywordAnalyzer(), IndexWriter.MaxFieldLength.UNLIMITED)) {
                    var doc = new Document();
                    doc.fields.Add(new NumericField("id", Libs.Lucene.Net.Document.Field.Store.YES, true).SetIntValue(entity.TflBatchId));
                    doc.fields.Add(new Libs.Lucene.Net.Document.Field("process", entity.ProcessName, Libs.Lucene.Net.Document.Field.Store.YES, Libs.Lucene.Net.Document.Field.Index.NOT_ANALYZED_NO_NORMS));
                    doc.fields.Add(new Libs.Lucene.Net.Document.Field("connection", input.Name, Libs.Lucene.Net.Document.Field.Store.YES, Libs.Lucene.Net.Document.Field.Index.NOT_ANALYZED_NO_NORMS));
                    doc.fields.Add(new Libs.Lucene.Net.Document.Field("entity", entity.Alias, Libs.Lucene.Net.Document.Field.Store.YES, Libs.Lucene.Net.Document.Field.Index.NOT_ANALYZED_NO_NORMS));
                    doc.fields.Add(new NumericField("updates", Libs.Lucene.Net.Document.Field.Store.YES, true).SetLongValue(entity.Updates));
                    doc.fields.Add(new NumericField("inserts", Libs.Lucene.Net.Document.Field.Store.YES, true).SetLongValue(entity.Inserts));
                    doc.fields.Add(new NumericField("deletes", Libs.Lucene.Net.Document.Field.Store.YES, true).SetLongValue(entity.Deletes));
                    doc.fields.Add(LuceneWriter.CreateField("version", versionType, new SearchType { Analyzer = "keyword" }, end));
                    doc.fields.Add(new Libs.Lucene.Net.Document.Field("version_type", versionType, Libs.Lucene.Net.Document.Field.Store.YES, Libs.Lucene.Net.Document.Field.Index.NOT_ANALYZED_NO_NORMS));
                    doc.fields.Add(new NumericField("tflupdate", Libs.Lucene.Net.Document.Field.Store.YES, true).SetLongValue(DateTime.UtcNow.Ticks));
                    writer.AddDocument(doc);
                    writer.Commit();
                    writer.Optimize();
                }
            }
        }

 public SqlServerEntityKeysExtractFromOutput(AbstractConnection connection, Entity entity)
     : base(connection) {
     _connection = connection;
     EntityName = entity.Name;
     _entity = entity;
     _fields = new List<string>(new FieldSqlWriter(entity.PrimaryKey).AddDeleted(entity).Alias(connection.L, connection.R).Keys()) { "TflKey" };
     _key = _entity.PrimaryKey;
 }

        public SqlKeysOverrideOperation(Entity entity, AbstractConnection connection)
            : base(connection) {
            _entity = entity;
            EntityName = entity.Name;
            _fields = entity.PrimaryKey.Aliases().ToArray();
            _length = _fields.Length;

            }

 public ElasticSearchEntityOutputKeysExtract(AbstractConnection connection, Entity entity) {
     _aliasTypes = entity.PrimaryKey.AliasTypes().ToList();
     if (entity.Version != null) {
         _aliasTypes.Add(new AliasType() { Alias = entity.Version.Alias, AliasLower = entity.Version.Alias.ToLower(), SimpleType = entity.Version.SimpleType });
     }
     _sourceInclude = _aliasTypes.Select(at => at.AliasLower).ToArray();
     _client = new ElasticSearchClientFactory().Create(connection, entity);
 }

        public ElasticSearchEntityExtract(AbstractConnection connection, Entity entity, IEnumerable<string> fields, bool correspondingKeys = false) {
            _entity = entity;
            _correspondingKeys = correspondingKeys;

            _fields = fields.Select(a => a.ToLower()).ToArray(); //for now
            _client = new ElasticSearchClientFactory().CreateNest(connection, entity);
            Name = string.Format("ElasticsearchEntityExtract ({0}:{1})", _client.Index, _client.Type);
        }

 public SqlEntityKeysExtractAllFromOutput(AbstractConnection connection, Entity entity)
     : base(connection) {
     _entity = entity;
     EntityName = entity.Name;
     _keys = _entity.Delete && _entity.IsMaster() ?
         new List<string>(entity.PrimaryKey.Aliases()) { "TflDeleted", "TflKey" } :
         new List<string>(entity.PrimaryKey.Aliases()) { "TflKey" };
 }

 public bool Check(AbstractConnection connection) {
     var fileInfo = new FileInfo(connection.File);
     if (!fileInfo.Exists) {
         _logger.Warn("File {0} does not exist.  Creating it.", fileInfo.Name);
         System.IO.File.Create(fileInfo.FullName).Dispose();
     }
     return true;
 }

 public SolrEntityOutputKeysExtract(AbstractConnection connection, Entity entity) {
     _entity = entity;
     _aliasTypes = _entity.PrimaryKey.AliasTypes().ToList();
     if (_entity.Version != null) {
         _aliasTypes.Add(new AliasType() { Alias = _entity.Version.Alias, AliasLower = _entity.Version.Alias.ToLower(), SimpleType = _entity.Version.SimpleType });
     }
     _sourceInclude = _aliasTypes.Select(at => at.AliasLower).ToArray();
 }