TODO - reviewed no change needed? datatypes need reviewing 
Get schema for connection
@param {*} connection
15530
function getSchema(connection) {
  const schema_sql = connection.schema_sql
    ? connection.schema_sql
    : SCHEMA_SQL_INFORMATION_SCHEMA;
  return runQuery(schema_sql, connection).then(queryResult =>
    formatSchemaQueryResults(queryResult)
  );
}
1290333
private function insertRow($tableName, Connection $connection, array $data, Closure $getColumnType): void
    {
        $types        = [];
        $quotedFields = [];
        foreach ($data as $column => $value) {
            $name                = $connection->quoteIdentifier($column);
            $quotedFields[$name] = $value;
            $types[$name]        = $getColumnType($column);
        }

        try {
            $result = $connection->insert($tableName, $quotedFields, $types);
            assert($result !== false, 'Insert failed');
        } /** @noinspection PhpRedundantCatchClauseInspection */ catch (UniqueConstraintViolationException $e) {
            // ignore non-unique records
        }
    }
545503
def init_copy(self, connection):
        
        if not self.does_schema_exist(connection):
            logger.info("Creating schema for %s", self.table)
            self.create_schema(connection)

        if not self.does_table_exist(connection):
            logger.info("Creating table %s", self.table)
            self.create_table(connection)

        if self.enable_metadata_columns:
            self._add_metadata_columns(connection)

        if self.do_truncate_table:
            logger.info("Truncating table %s", self.table)
            self.truncate_table(connection)

        if self.do_prune():
            logger.info("Removing %s older than %s from %s", self.prune_column, self.prune_date, self.prune_table)
            self.prune(connection)
837387
def db_type(self, connection):
        
        conn_module = type(connection).__module__
        if "mysql" in conn_module:
            return "bigint AUTO_INCREMENT"
        elif "postgres" in conn_module:
            return "bigserial"
        return super(BigAutoField, self).db_type(connection)
702803
function (connection, attempt) {
  var clonedAttempt = EJSON.clone(attempt);
  clonedAttempt.connection = connection;
  return clonedAttempt;
}
485493
public void connectClient(Connection connection) {
    threadContext.checkThread();

    // Note we do not use method references here because the "state" variable changes over time.
    // We have to use lambdas to ensure the request handler points to the current state.
    connection.handler(RegisterRequest.class, (Function<RegisterRequest, CompletableFuture<RegisterResponse>>) request -> state.register(request));
    connection.handler(ConnectRequest.class, (Function<ConnectRequest, CompletableFuture<ConnectResponse>>) request -> state.connect(request, connection));
    connection.handler(KeepAliveRequest.class, (Function<KeepAliveRequest, CompletableFuture<KeepAliveResponse>>) request -> state.keepAlive(request));
    connection.handler(UnregisterRequest.class, (Function<UnregisterRequest, CompletableFuture<UnregisterResponse>>) request -> state.unregister(request));
    connection.handler(ResetRequest.class, (Consumer<ResetRequest>) request -> state.reset(request));
    connection.handler(CommandRequest.class, (Function<CommandRequest, CompletableFuture<CommandResponse>>) request -> state.command(request));
    connection.handler(QueryRequest.class, (Function<QueryRequest, CompletableFuture<QueryResponse>>) request -> state.query(request));

    connection.onClose(stateMachine.executor().context().sessions()::unregisterConnection);
  }
43067
function getSchema(connection) {
  connection.maxRows = Number.MAX_SAFE_INTEGER;
  const driver = drivers[connection.driver];
  return driver.getSchema(connection);
}
458181
function (connection) {
        delete Transaction.connections[connection.transactionId];
        connection.transactionId = null;
        connection.transactionConfig = null;
    }
697472
function runQuery(query, connection) {
  const params = {
    host: connection.host,
    port: connection.port ? connection.port : 5433,
    user: connection.username,
    password: connection.password,
    database: connection.database
  };

  return new Promise((resolve, reject) => {
    const client = vertica.connect(params, function(err) {
      if (err) {
        client.disconnect();
        return reject(err);
      }

      let incomplete = false;
      const rows = [];
      let finished = false;
      let columnNames = [];

      const verticaQuery = client.query(query);

      verticaQuery.on('fields', fields => {
        columnNames = fields.map(field => field.name);
      });

      verticaQuery.on('row', function(row) {
        if (rows.length < connection.maxRows) {
          const resultRow = {};
          row.forEach((value, index) => {
            resultRow[columnNames[index]] = value;
          });
          return rows.push(resultRow);
        }
        if (!finished) {
          finished = true;
          client.disconnect();
          incomplete = true;
          return resolve({ rows, incomplete });
        }
      });

      verticaQuery.on('end', function() {
        if (!finished) {
          finished = true;
          client.disconnect();
          return resolve({ rows, incomplete });
        }
      });

      verticaQuery.on('error', function(err) {
        if (!finished) {
          finished = true;
          client.disconnect();
          return reject(err);
        }
      });
    });
  });
}
791482
function(connection, collection, functionName) {
      var args = Array.prototype.slice.call(arguments, 3);
      return connections[connection].runFunction(functionName, args);
    }
680306
public function initConnection(Connection $connection)
    {
        /** @var $pdo PDO */
        $pdo = $connection->getWrappedConnection();
        $pdo->setAttribute(PDO::ATTR_EMULATE_PREPARES, true);
        $params = $connection->getParams();
        if (!isset($params['charset'])) {
            throw new RuntimeException("No connection charset specified");
        }
        $pdo->exec('SET NAMES ' . $pdo->quote($params['charset']));
    }
447245
protected function createDatabaseSchema(Connection $connection, DatabaseSchemaInterface $schema): void
    {
        try {
            $this->createDatabaseTables($connection, $schema);
            $this->createDatabaseViews($connection, $schema);
        } catch (DBALException $exception) {
            if ($connection->isConnected() === true) {
                $this->removeDatabaseSchema($connection, $schema);
            }

            throw $exception;
        }
    }
1284561
function schema (args) {
    // get args object
    args = requireValidOptionalObject(args)
    // get colunmn and index information from database
    return Promise.all([
        this.mysql().query(sql.describe(this)),
        this.mysql().query(sql.showCreateTable(this)),
        this.mysql().query(sql.showIndexes(this)),
    ])
    // build schema from database response
    .then(res => {
        // get columns and indexes from describe and show index queries
        var columns = res[0][0]
        var createTable = res[1][0]
        var indexes = res[2][0]
        // schema data
        var schema = {
            columns: {},
            indexes: [],
        }
        // get first and only create table record
        createTable = _.get(createTable, "[0]['Create Table']", '')
        // get engine and charset from create table sql
        var engine = createTable.match(/ENGINE=(\w+)/)
        schema.engine = engine && engine[1] || ''
        var charset = createTable.match(/DEFAULT CHARSET=(\w+)/)
        schema.charset = charset && charset[1] || ''
        // build column data
        _.each(columns, dbColumn => {
            var column = schema.columns[dbColumn.Field] = {}
            // check unsigned
            if (dbColumn.Type.match(/unsigned$/)) {
                column.unsigned = true
                dbColumn.Type = dbColumn.Type.replace(/ unsigned/, '')
            }
            // char id column
            if (dbColumn.Type === 'char(32)') {
                // model must not be set for binary ids
                assert.ok(this.binaryIds === false, 'binaryIds must be disabled for char id columns')
                // set type
                column.type = 'id'
            }
            // binary id column
            else if (dbColumn.Type === 'binary(16)') {
                // model must be set for binary ids
                assert.ok(this.binaryIds === true, 'binaryIds must be enabled for binary id columns')
                // set type
                column.type = 'id'
            }
            // all other column types
            else  {
                // require valid type
                assert.ok(sql.columnTypesInverse[dbColumn.Type], 'unrecognized column type '+dbColumn.Type)
                // map database to internal type
                column.type = sql.columnTypesInverse[dbColumn.Type]
            }
            // is column nullable
            if (dbColumn.Null === 'NO') {
                column.null = false
            }
            // set default if not null
            if (dbColumn.Default) {
                column.default = dbColumn.Default
                // convert boolean
                if (column.type === 'boolean') {
                    column.default = column.default === '1' ? true : false
                }
            }
        })
        // map of indexes by key name since keys may cover multiple columns
        var indexesByKey = {}
        // build map of indexes by key name
        _.each(indexes, dbIndex => {
            // create entry for index if it does not already exist
            if (!indexesByKey[dbIndex.Key_name]) {
                indexesByKey[dbIndex.Key_name] = {
                    columns: {}
                }
            }
            // get entry for index
            var index = indexesByKey[dbIndex.Key_name]
            // add column with sequence in index
            index.columns[dbIndex.Column_name] = parseInt(dbIndex.Seq_in_index)
            // add unique flag if true
            if (dbIndex.Non_unique === '0') {
                index.unique = true
            }
        })
        // add indexes to schema
        _.each(indexesByKey, (index, name) => {
            var indexColumns = _.keys(index.columns)
            // multi-column index
            if (indexColumns.length > 1) {
                // sort index columns by the Seq_in_index value - these may
                // always be in order so this may not be needed
                indexColumns = _.sortBy(indexColumns, indexColumn => index.columns[indexColumn])
                // create schema index entry
                var schemaIndex = {
                    columns: indexColumns,
                }
                // set unique flag
                if (index.unique) {
                    schemaIndex.unique = true
                }
                // add index to schema
                schema.indexes.push(schemaIndex)
            }
            // single column index
            else {
                // get first and only column
                var indexColumn = indexColumns[0]
                // PRIMARY KEY
                if (name === 'PRIMARY') {
                    schema.columns[indexColumn].primary = true
                }
                // unique index
                else if (index.unique) {
                    schema.columns[indexColumn].unique = true
                }
                // non-unique index
                else {
                    schema.columns[indexColumn].index = true
                }
            }
        })
        // resolve with schema
        return schema
    })
    .catch(err => {
        // reject on anything besides a table does not exist error
        // if table does not exist resolve with undefined
        if (err.errno !== 1146) {
            throw err
        }
    })
}
744815
function(modelJs) {
    let web = this;
    let modelName = modelJs.name;
    
    if (!modelJs.schema) {
      throw new Error(modelName + '.schema not found.');
    }

    let collectionName = undefined;
    if (modelJs.parentModel) {
      
      let parentModel = web.includeModel(modelJs.parentModel);
      let parentModelJs = parentModel.getModelDictionary();
      collectionName = parentModel.collection.name;

      modelJs.schema = extend(parentModelJs.schema, modelJs.schema);
      
      let origSchema = modelJs.initSchema;
      modelJs.initSchema = [];
      if (origSchema) {
        modelJs.initSchema.push(origSchema);
      }

      if (parentModelJs.initSchema) {
        modelJs.initSchema.push(parentModelJs.initSchema);
      }

      modelJs.options = extend(parentModel.options || {}, modelJs.options);

      if (console.isDebug) {
        console.debug('Model %s has a parent %s', modelName, modelJs.parentModel);
      }

    }

    let conn;

    let getModelConnection = function(modelJs) {
      if (modelJs.connection) {
        return modelJs.connection
      }
      
      if (modelJs.parentModel) {
        return getModelConnection(web.include(modelJs.parentModel));
      }
    }

    let modelConn = getModelConnection(modelJs);
    if (modelConn) {
      conn = this.connections[modelConn];
      if (web.conf.isDebug) {
        console.debug("Found model conn: ", modelJs.name, modelConn);
      }
    } else if (web.connections.mainDb) {
      conn = web.connections.mainDb; 
    } else {
    for (let i in this.connections) {
        //get the first connection
        conn = this.connections[i];
        break;
      }
    }

    if (!conn) {
      console.warn('No defined DB. Check your configuration.');
      return;
    }
    
    let schema = new Schema(modelJs.schema, modelJs.options);

    //TODO: executing schemas of children are not a good idea
    if (modelJs.initSchema) {
      if (modelJs.initSchema instanceof Array) {
        for (let i in modelJs.initSchema) {
          let mySchema = modelJs.initSchema[i];
          
          mySchema(schema);
        }
      } else {
        //console.debug('[%s] Executing normal initSchema %s', modelJs.name ,modelJs.initSchema);
        //fixed bug where schemas are execd twice
        if (!modelJs.initSchema.execd) {
          modelJs.initSchema(schema);
        }
      }
      
    }

    let model = conn.model(modelName, schema, collectionName);
    if (console.isDebug) {
      console.debug("Loaded model for the first time: " + modelName)
    }

    web.modelCache[modelName] = model;
    model.getModelDictionary = function() {
      return modelJs;
    }        
    return model;
  }
1289105
public function updateSchema($event = null, $connection = null) {
		if (empty($event) || !is_array($event)) {
			return false;
		}

		if (empty($connection)) {
			$connection = 'default';
		}

		$errors = Hash::get($event, 'errors');
		if (!empty($errors)) {
			return false;
		}

		$table = Hash::get($event, 'create');
		if (empty($table)) {
			$table = Hash::get($event, 'update');
		}
		if (empty($table)) {
			return false;
		}

		$name = 'CakeLdap';
		$plugin = 'CakeLdap';
		$Schema = new CakeSchema(compact('connection', 'plugin', 'name'));
		$oldSchema = $Schema->load();
		if ($oldSchema === false) {
			return false;
		}

		$newSchema = $Schema->load();
		if (!isset($newSchema->tables[$table]) || empty($newSchema->tables[$table])) {
			return false;
		}

		$modelConfigSync = ClassRegistry::init('CakeLdap.ConfigSync');
		$fieldsDb = $modelConfigSync->getListFieldsDb();
		$fieldsLdap = $modelConfigSync->getListFieldsLdap();
		if (empty($fieldsDb)) {
			return false;
		}

		$fieldsInt = [
			'indexes',
			'tableParameters'
		];

		$newSchema->tables = array_intersect_key($newSchema->tables, [$table => null]);
		switch ($table) {
			case 'departments':
				if (in_array('department_id', $fieldsDb)) {
					$deleteDepartments = $modelConfigSync->getFlagDeleteDepartments();
					if (!$deleteDepartments) {
						return true;
					}

					unset($newSchema->tables[$table]['block']);
				} else {
					unset($newSchema->tables[$table]);
				}
				break;
			/*
			case 'memberofs':
				if (in_array(CAKE_LDAP_LDAP_ATTRIBUTE_MEMBER_OF, $fieldsLdap))
					return true;

				unset($newSchema->tables[$table]);
			break;
			*/
			case 'othertelephones':
				if (in_array(CAKE_LDAP_LDAP_ATTRIBUTE_OTHER_TELEPHONE_NUMBER, $fieldsLdap)) {
					return true;
				}

				unset($newSchema->tables[$table]);
				break;
			case 'othermobiles':
				if (in_array(CAKE_LDAP_LDAP_ATTRIBUTE_OTHER_MOBILE_TELEPHONE_NUMBER, $fieldsLdap)) {
					return true;
				}

				unset($newSchema->tables[$table]);
				break;
			case 'subordinates':
				if (in_array('manager_id', $fieldsDb)) {
					return true;
				}

				unset($newSchema->tables[$table]);
				break;
			case 'employees':
				$fieldsSchema = array_merge($fieldsDb, $fieldsInt);
				$newSchema->tables[$table] = array_intersect_key($newSchema->tables[$table], array_flip($fieldsSchema));
				break;
			default:
				return false;
		}

		$compareNew = $Schema->compare($oldSchema, $newSchema);
		$compareOld = $Schema->compare($newSchema, $oldSchema);
		$compareDataNew = Hash::get($compareNew, $table);
		$compareDataOld = Hash::get($compareOld, $table);

		$contents = null;
		$ds = ConnectionManager::getDataSource($Schema->connection);
		$ds->cacheSources = false;
		if (!empty($compareDataNew)) {
			$contents = [$table => $ds->alterSchema([$table => $compareDataNew], $table)];
		} elseif (!empty($compareDataOld)) {
			$contents = [$table => $ds->dropSchema($oldSchema, $table)];
		}
		if (empty($contents)) {
			return true;
		}

		$result = true;
		foreach ($contents as $table => $sql) {
			if (empty($sql)) {
				continue;
			}

			try {
				$ds->execute($sql);
			} catch (PDOException $e) {
				$result = false;
			}
		}

		return $result;
	}
1350273
function validateDataSource(connections, dataSource, cb) {
  var dataSourceToValidate = _.clone(dataSource);
  var DataSource = models.get(connections.mongooseConnection, models.MODELNAMES.DATA_SOURCE);

  //Expect The Data Source To Have Data
  var failure = validate(dataSourceToValidate).has(CONSTANTS.DATA_SOURCE_DATA);

  if (failure) {
    return cb(buildErrorResponse({
      error: new Error("No Data Passed To Validate"),
      code: ERROR_CODES.FH_FORMS_INVALID_PARAMETERS
    }));
  }

  //Generate A Hash
  var hash = misc.generateHash(dataSourceToValidate.data);

  //Populate A Cache Object
  //LastRefreshed is populated just for the cache mongoose schema
  var cache = {
    dataHash: hash,
    currentStatus: {
      status: "ok"
    },
    data: dataSourceToValidate.data,
    lastRefreshed: new Date().getTime(),
    updateTimestamp: new Date().getTime()
  };

  dataSourceToValidate.cache = [cache];

  //Create A New Mongoose. Note that this document is never saved. It is useful for consistent validation
  var testDataSource = new DataSource(dataSourceToValidate);
  //Validating Without Saving
  testDataSource.validate(function(err) {
    var valid = err ? false : true;
    var dataSourceJSON = testDataSource.toJSON();

    dataSourceJSON = processDataSourceResponse(dataSourceJSON);

    dataSourceJSON = _.omit(dataSourceJSON, "lastRefreshed", "currentStatus", "updateTimestamp", CONSTANTS.DATA_SOURCE_ID);

    dataSourceJSON.validationResult = {
      valid: valid,
      message: valid ? "Data Source Is Valid" : "Invalid Data Source Update Data."
    };

    return cb(undefined, dataSourceJSON);
  });
}
1133239
function getMongooseConnection(mongoDbUrl, logger, cb) {
  logger.debug("creating mongoose connection for data_source_update job", {mongoDbUrl: mongoDbUrl});
  var mongooseConnection = mongoose.createConnection(mongoDbUrl);
  return cb(undefined, mongooseConnection);
}