func (s *Storage) RemoveCollection(ctx context.Context, dbname, shardinstance, collectionname string) error {
	tableRemoveQuery := fmt.Sprintf(removeTableTemplate, shardinstance, collectionname)
	if _, err := s.dbMap[dbname].Query(tableRemoveQuery); err != nil {
		return fmt.Errorf("Unable to run tableRemoveQuery%s: %v", collectionname, err)
	}
	return nil
}
TODO: use db listing to remove things
TODO: remove indexes on removal
package pgstorage
import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"github.com/jacksontj/dataman/datamantype"
	"github.com/jacksontj/dataman/storagenode/metadata"
	"github.com/sirupsen/logrus"
)
func fieldToSchema(field *metadata.CollectionField) (string, error) {
	fieldStr := ""
	switch field.FieldType.DatamanType {
	case datamantype.JSON:
		fieldStr += "\"" + field.Name + "\" jsonb"
	case datamantype.Document:
		fieldStr += "\"" + field.Name + "\" jsonb"
	case datamantype.String:
		fieldStr += "\"" + field.Name + "\" character varying(255)"
	case datamantype.Text:
		fieldStr += "\"" + field.Name + "\" text"
	case datamantype.Int:
		fieldStr += "\"" + field.Name + "\" int8"
	case datamantype.Float:
		fieldStr += "\"" + field.Name + "\" double precision"
	case datamantype.Serial:
		fieldStr += "\"" + field.Name + "\" serial8"
	case datamantype.Bool:
		fieldStr += "\"" + field.Name + "\" bool"
	case datamantype.DateTime:
		fieldStr += "\"" + field.Name + "\" timestamp without time zone"
	default:
		return "", fmt.Errorf("Unknown field type: %v", field.Type)
	}
	if field.NotNull {
		fieldStr += " NOT NULL"
	}
	if field.Default != nil {
		fieldStr += fmt.Sprintf(" DEFAULT %v", field.Default)
	}
	return fieldStr, nil
}
func (s *Storage) ListDatabase(ctx context.Context) []*metadata.Database {
	if dbRecords, err := DoQuery(ctx, s.db, "SELECT datname FROM pg_database WHERE datistemplate = false", nil); err == nil {
		dbs := make([]*metadata.Database, len(dbRecords))
		for i, dbRecord := range dbRecords {
			dbs[i] = s.GetDatabase(ctx, string(dbRecord["datname"].([]byte)))
		}
		return dbs
	}
	return nil
}
func (s *Storage) GetDatabase(ctx context.Context, dbname string) *metadata.Database {
	dbQuery := fmt.Sprintf("SELECT datname FROM pg_database WHERE datistemplate = false AND datname='%s'", dbname)
	dbRecords, err := DoQuery(ctx, s.db, dbQuery, nil)
	if err != nil || len(dbRecords) != 1 {
		return nil
	}
	database := metadata.NewDatabase(dbname)
	database.ProvisionState = metadata.Active
	shardInstances := s.ListShardInstance(ctx, dbname)
	if shardInstances != nil {
		for _, shardInstance := range shardInstances {
			database.ShardInstances[shardInstance.Name] = shardInstance
		}
	}
	return database
}
func (s *Storage) AddDatabase(ctx context.Context, db *metadata.Database) error {
	if _, err := DoQuery(ctx, s.db, fmt.Sprintf("CREATE DATABASE \"%s\"", db.Name), nil); err != nil {
		return fmt.Errorf("Unable to create database: %v", err)
	}
	dbConn, err := sql.Open("postgres", s.config.pgStringForDB(db.Name))
	if err != nil {
		return fmt.Errorf("Unable to open db connection: %v", err)
	}
	s.dbMap[db.Name] = dbConn
	return nil
}
const dropDatabaseTemplate = `DROP DATABASE IF EXISTS "%s";`
func (s *Storage) RemoveDatabase(ctx context.Context, dbname string) error {
	if conn, ok := s.dbMap[dbname]; ok {
		if err := conn.Close(); err != nil {
			return fmt.Errorf("Unable to close DB connection during RemoveDatabase: %v", err)
		}
	}
	_, err := DoQuery(ctx, s.db, fmt.Sprintf(`SELECT pg_terminate_backend(pg_stat_activity.pid)
        FROM pg_stat_activity
        WHERE pg_stat_activity.datname = '%s';`, dbname), nil)
	if err != nil {
		return fmt.Errorf("Unable to close open connections: %v", err)
	}
	if _, err := DoQuery(ctx, s.db, fmt.Sprintf(dropDatabaseTemplate, dbname), nil); err != nil {
		return fmt.Errorf("Unable to drop db: %v", err)
	}
	return nil
}
func (s *Storage) ListShardInstance(ctx context.Context, dbname string) []*metadata.ShardInstance {
	schemas, err := DoQuery(ctx, s.getDB(dbname), "SELECT * from information_schema.schemata", nil)
	if err != nil {
		logrus.Fatalf("Unable to get shard list for db %s: %v", dbname, err)
	}
	shardInstances := make([]*metadata.ShardInstance, 0)
	for _, schemaRecord := range schemas {
		schemaName := schemaRecord["schema_name"].(string)
		if strings.HasPrefix(schemaName, "pg_") {
			continue
		}
		switch schemaName {
		case "information_schema":
			continue
		default:
			shardInstance := metadata.NewShardInstance(schemaName)
			shardInstance.ProvisionState = metadata.Active
			shardInstance.Count = 1
			shardInstance.Instance = 1
			collections := s.ListCollection(ctx, dbname, schemaName)
			for _, collection := range collections {
				shardInstance.Collections[collection.Name] = collection
			}
			shardInstances = append(shardInstances, shardInstance)
		}
	}
	return shardInstances
}
func (s *Storage) GetShardInstance(ctx context.Context, dbname, shardinstance string) *metadata.ShardInstance {
	shardInstances := s.ListShardInstance(ctx, dbname)
	if shardInstances == nil {
		return nil
	}
	for _, shardInstance := range shardInstances {
		if shardInstance.Name == shardinstance {
			return shardInstance
		}
	}
	return nil
}
func (s *Storage) AddShardInstance(ctx context.Context, db *metadata.Database, shardInstance *metadata.ShardInstance) error {
	if _, err := DoQuery(ctx, s.getDB(db.Name), fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS \"%s\"", shardInstance.Name), nil); err != nil {
		return fmt.Errorf("Unable to create schema: %v", err)
	}
	return nil
}
func (s *Storage) RemoveShardInstance(ctx context.Context, dbname, shardInstance string) error {
	if _, err := DoQuery(ctx, s.getDB(dbname), fmt.Sprintf("DROP SCHEMA IF EXISTS \"%s\"", shardInstance), nil); err != nil {
		return fmt.Errorf("Unable to drop schema: %v", err)
	}
	return nil
}
var listRelationQuery = `
select c.constraint_name
    , x.table_schema as schema_name
    , x.table_name
    , x.column_name
    , y.table_schema as foreign_schema_name
    , y.table_name as foreign_table_name
    , y.column_name as foreign_column_name
from information_schema.referential_constraints c
join information_schema.key_column_usage x
    on x.constraint_name = c.constraint_name
join information_schema.key_column_usage y
    on y.ordinal_position = x.position_in_unique_constraint
    and y.constraint_name = c.unique_constraint_name
WHERE x.table_schema = '%s' AND x.table_name = '%s'
`
func (s *Storage) ListCollection(ctx context.Context, dbname, shardinstance string) []*metadata.Collection {
	query := fmt.Sprintf("SELECT table_name FROM information_schema.tables WHERE table_schema='%s' ORDER BY table_schema,table_name;", shardinstance)
	tables, err := DoQuery(ctx, s.getDB(dbname), query, nil)
	if err != nil {
		logrus.Fatalf("Unable to get table list for db %s: %v", dbname, err)
	}
	collections := make([]*metadata.Collection, len(tables))
	for i, tableEntry := range tables {
		tableName := tableEntry["table_name"].(string)
		collection := metadata.NewCollection(tableName)
		collection.ProvisionState = metadata.Active
		collection.Fields = make(map[string]*metadata.CollectionField)
		for _, field := range s.ListCollectionField(ctx, dbname, shardinstance, collection.Name) {
			collection.Fields[field.Name] = field
		}
		collection.Indexes = make(map[string]*metadata.CollectionIndex)
		for _, index := range s.ListCollectionIndex(ctx, dbname, shardinstance, collection.Name) {
			collection.Indexes[index.Name] = index
		}
		collections[i] = collection
	}
	return collections
}
func (s *Storage) GetCollection(ctx context.Context, dbname, shardinstance, collectionname string) *metadata.Collection {
	collections := s.ListCollection(ctx, dbname, shardinstance)
	if collections == nil {
		return nil
	}
	for _, collection := range collections {
		if collection.Name == collectionname {
			return collection
		}
	}
	return nil
}
const addTableTemplate = `CREATE TABLE "%s".%s()`
func (s *Storage) AddCollection(ctx context.Context, db *metadata.Database, shardInstance *metadata.ShardInstance, collection *metadata.Collection) error {
	tableAddQuery := fmt.Sprintf(addTableTemplate, shardInstance.Name, collection.Name)
	if _, err := DoQuery(ctx, s.getDB(db.Name), tableAddQuery, nil); err != nil {
		return fmt.Errorf("Unable to add collection %s: %v", collection.Name, err)
	}
	return nil
}
const removeTableTemplate = `DROP TABLE %s.%s`