func toSchemaKeys(conv *internal.Conv, table string, s []string) (l []schema.Key) {
	for _, k := range s {
		l = append(l, schema.Key{Column: k})
	}
	return l
}
toSchemaKeys converts a string list of PostgreSQL primary keys to
schema primary keys.
package postgres
import (
	"fmt"
	"reflect"
	"strconv"
	"strings"
	"time"
	pg_query "github.com/pganalyze/pg_query_go/v2"
	"github.com/cloudspannerecosystem/harbourbridge/internal"
	"github.com/cloudspannerecosystem/harbourbridge/schema"
	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
)
type DbDumpImpl struct{}
type copyOrInsert struct {
	stmt  stmtType
	table string
	cols  []string
	rows  [][]string  
}
type stmtType int
const (
	copyFrom stmtType = iota
	insert
)
func (ddi DbDumpImpl) GetToDdl() common.ToDdl {
	return ToDdlImpl{}
}
func (ddi DbDumpImpl) ProcessDump(conv *internal.Conv, r *internal.Reader) error {
	return processPgDump(conv, r)
}
func processPgDump(conv *internal.Conv, r *internal.Reader) error {
	for {
		startLine := r.LineNumber
		startOffset := r.Offset
		b, stmts, err := readAndParseChunk(conv, r)
		if err != nil {
			return err
		}
		ci := processStatements(conv, stmts)
		internal.VerbosePrintf("Parsed SQL command at line=%d/fpos=%d: %d stmts (%d lines, %d bytes) ci=%v\n", startLine, startOffset, len(stmts), r.LineNumber-startLine, len(b), ci != nil)
		if ci != nil {
			switch ci.stmt {
			case copyFrom:
				processCopyBlock(conv, ci.table, ci.cols, r)
			case insert:
				for _, vals := range ci.rows {
					if len(ci.cols) == 0 {
						ProcessDataRow(conv, ci.table, conv.SrcSchema[ci.table].ColNames, vals)
					} else {
						ProcessDataRow(conv, ci.table, ci.cols, vals)
					}
				}
			}
		}
		if r.EOF {
			break
		}
	}
	return nil
}
func readAndParseChunk(conv *internal.Conv, r *internal.Reader) ([]byte, []*pg_query.RawStmt, error) {
	var l [][]byte
	for {
		b := r.ReadLine()
		l = append(l, b)
		if strings.Contains(string(b), ";") || r.EOF {
			n := 0
			for i := range l {
				n += len(l[i])
			}
			s := make([]byte, n)
			n = 0
			for i := range l {
				n += copy(s[n:], l[i])
			}
			tree, err := pg_query.Parse(string(s))
			if err == nil {
				return s, tree.Stmts, nil
			}
			conv.Stats.Reparsed++
		}
		if r.EOF {
			return nil, nil, fmt.Errorf("error parsing last %d line(s) of input", len(l))
		}
	}
}
func processCopyBlock(conv *internal.Conv, srcTable string, srcCols []string, r *internal.Reader) {
	internal.VerbosePrintf("Parsing COPY-FROM stdin block starting at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
	for {
		b := r.ReadLine()
		if string(b) == "\\.\n" || string(b) == "\\.\r\n" {
			internal.VerbosePrintf("Parsed COPY-FROM stdin block ending at line=%d/fpos=%d\n", r.LineNumber, r.Offset)
			return
		}
		if r.EOF {
			conv.Unexpected("Reached eof while parsing copy-block")
			return
		}
		conv.StatsAddRow(srcTable, conv.SchemaMode())
		if !conv.DataMode() {
			continue
		}
		s := strings.ReplaceAll(string(b), `\\`, `\`)
		ProcessDataRow(conv, srcTable, srcCols, strings.Split(strings.Trim(s, "\r\n"), "\t"))
	}
}
func processStatements(conv *internal.Conv, rawStmts []*pg_query.RawStmt) *copyOrInsert {
	for i, rawStmt := range rawStmts {
		node := rawStmt.Stmt
		switch n := node.GetNode().(type) {
		case *pg_query.Node_AlterTableStmt:
			if conv.SchemaMode() {
				processAlterTableStmt(conv, n.AlterTableStmt)
			}
		case *pg_query.Node_CopyStmt:
			if i != len(rawStmts)-1 {
				conv.Unexpected("CopyFrom is not the last statement in batch: ignoring following statements")
				conv.ErrorInStatement(printNodeType(n.CopyStmt))
			}
			return processCopyStmt(conv, n.CopyStmt)
		case *pg_query.Node_CreateStmt:
			if conv.SchemaMode() {
				processCreateStmt(conv, n.CreateStmt)
			}
		case *pg_query.Node_InsertStmt:
			return processInsertStmt(conv, n.InsertStmt)
		case *pg_query.Node_VariableSetStmt:
			if conv.SchemaMode() {
				processVariableSetStmt(conv, n.VariableSetStmt)
			}
		case *pg_query.Node_IndexStmt:
			if conv.SchemaMode() {
				processIndexStmt(conv, n.IndexStmt)
			}
		default:
			conv.SkipStatement(printNodeType(n))
		}
	}
	return nil
}
func processIndexStmt(conv *internal.Conv, n *pg_query.IndexStmt) {
	if n.Relation == nil {
		logStmtError(conv, n, fmt.Errorf("cannot process index statement with nil relation"))
		return
	}
	tableName, err := getTableName(conv, n.Relation)
	if err != nil {
		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
		return
	}
	if ctable, ok := conv.SrcSchema[tableName]; ok {
		ctable.Indexes = append(ctable.Indexes, schema.Index{
			Name:   n.Idxname,
			Unique: n.Unique,
			Keys:   toIndexKeys(conv, n.Idxname, n.IndexParams),
		})
		conv.SrcSchema[tableName] = ctable
	} else {
		conv.Unexpected(fmt.Sprintf("Table %s not found while processing index statement", tableName))
		conv.SkipStatement(printNodeType(n))
	}
}
func processAlterTableStmt(conv *internal.Conv, n *pg_query.AlterTableStmt) {
	if n.Relation == nil {
		logStmtError(conv, n, fmt.Errorf("relation is nil"))
		return
	}
	table, err := getTableName(conv, n.Relation)
	if err != nil {
		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
		return
	}
	if _, ok := conv.SrcSchema[table]; ok {
		for _, i := range n.Cmds {
			cmd := i.GetNode()
			switch t := cmd.(type) {
			case *pg_query.Node_AlterTableCmd:
				a := t.AlterTableCmd
				switch {
				case a.Subtype == pg_query.AlterTableType_AT_SetNotNull && a.Name != "":
					c := constraint{ct: pg_query.ConstrType_CONSTR_NOTNULL, cols: []string{a.Name}}
					updateSchema(conv, table, []constraint{c}, "ALTER TABLE")
					conv.SchemaStatement(strings.Join([]string{printNodeType(n), printNodeType(t)}, "."))
				case a.Subtype == pg_query.AlterTableType_AT_AddConstraint && a.Def != nil:
					switch at := a.Def.GetNode().(type) {
					case *pg_query.Node_Constraint:
						updateSchema(conv, table, extractConstraints(conv, printNodeType(n), table, []*pg_query.Node{a.Def}), "ALTER TABLE")
						conv.SchemaStatement(strings.Join([]string{printNodeType(n), printNodeType(t), printNodeType(at)}, "."))
					default:
						conv.SkipStatement(strings.Join([]string{printNodeType(n), printNodeType(t), printNodeType(at)}, "."))
					}
				default:
					conv.SkipStatement(strings.Join([]string{printNodeType(n), printNodeType(t)}, "."))
				}
			default:
				conv.SkipStatement(strings.Join([]string{printNodeType(n), printNodeType(t)}, "."))
			}
		}
	} else {
		conv.SkipStatement(printNodeType(n))
		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
	}
}
func processCreateStmt(conv *internal.Conv, n *pg_query.CreateStmt) {
	var colNames []string
	colDef := make(map[string]schema.Column)
	if n.Relation == nil {
		logStmtError(conv, n, fmt.Errorf("relation is nil"))
		return
	}
	table, err := getTableName(conv, n.Relation)
	if err != nil {
		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
		return
	}
	if len(n.InhRelations) > 0 {
		conv.SkipStatement(printNodeType(n))
		conv.Unexpected(fmt.Sprintf("Found inherited table %s -- we do not currently handle inherited tables", table))
		internal.VerbosePrintf("Processing %v statement: table %s is inherited table", printNodeType(n), table)
		return
	}
	var constraints []constraint
	for _, te := range n.TableElts {
		switch te.GetNode().(type) {
		case *pg_query.Node_ColumnDef:
			name, col, cdConstraints, err := processColumn(conv, te.GetColumnDef(), table)
			if err != nil {
				logStmtError(conv, n, err)
				return
			}
			colNames = append(colNames, name)
			colDef[name] = col
			constraints = append(constraints, cdConstraints...)
		case *pg_query.Node_Constraint:
			constraints = append(constraints, extractConstraints(conv, printNodeType(n), table, []*pg_query.Node{te})...)
		default:
			conv.Unexpected(fmt.Sprintf("Found %s node while processing CreateStmt TableElts", printNodeType(te)))
		}
	}
	conv.SchemaStatement(printNodeType(n))
	conv.SrcSchema[table] = schema.Table{
		Name:     table,
		ColNames: colNames,
		ColDefs:  colDef}
	updateSchema(conv, table, constraints, "CREATE TABLE")
}
func processColumn(conv *internal.Conv, n *pg_query.ColumnDef, table string) (string, schema.Column, []constraint, error) {
	mods := getTypeMods(conv, n.TypeName.Typmods)
	if n.Colname == "" {
		return "", schema.Column{}, nil, fmt.Errorf("colname is empty string")
	}
	name := n.Colname
	tid, err := getTypeID(n.TypeName.Names)
	if err != nil {
		return "", schema.Column{}, nil, fmt.Errorf("can't get type id for %s: %w", name, err)
	}
	ty := schema.Type{
		Name:        tid,
		Mods:        mods,
		ArrayBounds: getArrayBounds(conv, n.TypeName.ArrayBounds)}
	return name, schema.Column{Name: name, Type: ty}, analyzeColDefConstraints(conv, printNodeType(n), table, n.Constraints, name), nil
}
func processInsertStmt(conv *internal.Conv, n *pg_query.InsertStmt) *copyOrInsert {
	if n.Relation == nil {
		logStmtError(conv, n, fmt.Errorf("relation is nil"))
		return nil
	}
	table, err := getTableName(conv, n.Relation)
	if err != nil {
		logStmtError(conv, n, fmt.Errorf("can't get table name: %w", err))
		return nil
	}
	if _, ok := conv.SrcSchema[table]; !ok {
		conv.SkipStatement(printNodeType(n))
		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
		return nil
	}
	conv.StatsAddRow(table, conv.SchemaMode())
	colNames, err := getCols(conv, table, n.Cols)
	if err != nil {
		logStmtError(conv, n, fmt.Errorf("can't get col name: %w", err))
		conv.StatsAddBadRow(table, conv.SchemaMode())
		return nil
	}
	switch sel := n.SelectStmt.GetNode().(type) {
	case *pg_query.Node_SelectStmt:
		rows := getRows(conv, sel.SelectStmt.ValuesLists, n)
		conv.DataStatement(printNodeType(sel))
		if conv.DataMode() {
			return &copyOrInsert{stmt: insert, table: table, cols: colNames, rows: rows}
		}
	default:
		conv.Unexpected(fmt.Sprintf("Found %s node while processing InsertStmt SelectStmt", printNodeType(sel)))
	}
	return nil
}
func processCopyStmt(conv *internal.Conv, n *pg_query.CopyStmt) *copyOrInsert {
	table := "BOGUS_COPY_FROM_TABLE"
	var err error
	if n.Relation != nil {
		table, err = getTableName(conv, n.Relation)
		if err != nil {
			conv.Unexpected(fmt.Sprintf("Processing %v statement: %s", printNodeType(n), err))
		}
	} else {
		logStmtError(conv, n, fmt.Errorf("relation is nil"))
	}
	if _, ok := conv.SrcSchema[table]; !ok {
		conv.SkipStatement(printNodeType(n))
		internal.VerbosePrintf("Processing %v statement: table %s not found", printNodeType(n), table)
		return &copyOrInsert{stmt: copyFrom, table: table, cols: []string{}}
	}
	var cols []string
	for _, a := range n.Attlist {
		s, err := getString(a)
		if err != nil {
			conv.Unexpected(fmt.Sprintf("Processing %v statement Attlist: %s", printNodeType(n), err))
			s = "BOGUS_COPY_FROM_COLUMN"
		}
		cols = append(cols, s)
	}
	conv.DataStatement(printNodeType(n))
	return &copyOrInsert{stmt: copyFrom, table: table, cols: cols}
}
func processVariableSetStmt(conv *internal.Conv, n *pg_query.VariableSetStmt) {
	if n.Name == "timezone" {
		if len(n.Args) == 1 {
			arg := n.Args[0]
			switch c := arg.GetNode().(type) {
			case *pg_query.Node_AConst:
				tz, err := getString(c.AConst.Val)
				if err != nil {
					logStmtError(conv, c, fmt.Errorf("can't get Arg: %w", err))
					return
				}
				loc, err := time.LoadLocation(tz)
				if err != nil {
					logStmtError(conv, c, err)
					return
				}
				conv.SetLocation(loc)
			default:
				logStmtError(conv, arg, fmt.Errorf("found %s node in Arg", printNodeType(c)))
				return
			}
		}
	}
}
func getTypeMods(conv *internal.Conv, t []*pg_query.Node) (l []int64) {
	for _, x := range t {
		switch t1 := x.GetNode().(type) {
		case *pg_query.Node_AConst:
			switch t2 := t1.AConst.Val.GetNode().(type) {
			case *pg_query.Node_Integer:
				l = append(l, int64(t2.Integer.Ival))
			default:
				conv.Unexpected(fmt.Sprintf("Found %s node while processing Typmods", printNodeType(t2)))
			}
		default:
			conv.Unexpected(fmt.Sprintf("Found %s node while processing Typmods", printNodeType(t1)))
		}
	}
	return l
}
func getArrayBounds(conv *internal.Conv, t []*pg_query.Node) (l []int64) {
	for _, x := range t {
		switch t := x.GetNode().(type) {
		case *pg_query.Node_Integer:
			l = append(l, int64(t.Integer.Ival))
		default:
			conv.Unexpected(fmt.Sprintf("Found %s node while processing ArrayBounds", printNodeType(x)))
		}
	}
	return l
}
func getTypeID(nodes []*pg_query.Node) (string, error) {
	var ids []string
	for _, node := range nodes {
		s, err := getString(node)
		if err != nil {
			return "", err
		}
		ids = append(ids, s)
	}
	if len(ids) > 1 && ids[0] == "pg_catalog" {
		ids = ids[1:]
	}
	return strings.Join(ids, "."), nil
}
func getTableName(conv *internal.Conv, n *pg_query.RangeVar) (string, error) {
	var l []string
	if n.Catalogname != "" {
		l = append(l, n.Catalogname)
	}
	if n.Schemaname != "" && n.Schemaname != "public" {  
		l = append(l, n.Schemaname)
	}
	if n.Relname == "" {
		return "", fmt.Errorf("relname is empty: can't build table name")
	}
	l = append(l, n.Relname)
	return strings.Join(l, "."), nil
}
type constraint struct {
	ct   pg_query.ConstrType
	cols []string
	name string  
	referCols  []string
	referTable string
}
func extractConstraints(conv *internal.Conv, stmtType, table string, l []*pg_query.Node) (cs []constraint) {
	for _, i := range l {
		switch d := i.GetNode().(type) {
		case *pg_query.Node_Constraint:
			c := d.Constraint
			var cols, referCols []string
			var referTable string
			var conName string
			switch c.Contype {
			case pg_query.ConstrType_CONSTR_FOREIGN:
				t, err := getTableName(conv, c.Pktable)
				if err != nil {
					conv.Unexpected(fmt.Sprintf("Processing %v statement: error processing constraints: %s", printNodeType(d), err.Error()))
					conv.ErrorInStatement(printNodeType(d))
					continue
				}
				referTable = t
				if c.Conname != "" {
					conName = c.Conname
				}
				for _, attr := range c.FkAttrs {
					k, err := getString(attr)
					if err != nil {
						conv.Unexpected(fmt.Sprintf("Processing %v statement: error processing constraints: %s", printNodeType(d), err.Error()))
						conv.ErrorInStatement(printNodeType(d))
						continue
					}
					cols = append(cols, k)
				}
				for _, attr := range c.PkAttrs {
					f, err := getString(attr)
					if err != nil {
						conv.Unexpected(fmt.Sprintf("Processing %v statement: error processing constraints: %s", printNodeType(d), err.Error()))
						conv.ErrorInStatement(printNodeType(d))
						continue
					}
					referCols = append(referCols, f)
				}
			default:
				if c.Conname != "" {
					conName = c.Conname
				}
				for _, key := range c.Keys {
					k, err := getString(key)
					if err != nil {
						conv.Unexpected(fmt.Sprintf("Processing %v statement: error processing constraints: %s", printNodeType(d), err.Error()))
						conv.ErrorInStatement(fmt.Sprintf("%v.%v", printNodeType(i), printNodeType(d)))
						continue
					}
					cols = append(cols, k)
				}
			}
			cs = append(cs, constraint{ct: c.Contype, cols: cols, name: conName, referCols: referCols, referTable: referTable})
		default:
			conv.Unexpected(fmt.Sprintf("Processing %v statement: found %s node while processing constraints\n", stmtType, printNodeType(d)))
		}
	}
	return cs
}
func analyzeColDefConstraints(conv *internal.Conv, stmtType, table string, l []*pg_query.Node, pgCol string) (cs []constraint) {
	for _, c := range extractConstraints(conv, stmtType, table, l) {
		if len(c.cols) != 0 {
			conv.Unexpected("ColumnDef constraint has keys")
		}
		c.cols = []string{pgCol}
		cs = append(cs, c)
	}
	return cs
}
func updateSchema(conv *internal.Conv, table string, cs []constraint, stmtType string) {
	for _, c := range cs {
		switch c.ct {
		case pg_query.ConstrType_CONSTR_PRIMARY:
			ct := conv.SrcSchema[table]
			checkEmpty(conv, ct.PrimaryKeys, stmtType)
			ct.PrimaryKeys = toSchemaKeys(conv, table, c.cols)  
			updateCols(pg_query.ConstrType_CONSTR_NOTNULL, c.cols, ct.ColDefs)
			conv.SrcSchema[table] = ct
		case pg_query.ConstrType_CONSTR_FOREIGN:
			ct := conv.SrcSchema[table]
			ct.ForeignKeys = append(ct.ForeignKeys, toForeignKeys(c))  
			conv.SrcSchema[table] = ct
		case pg_query.ConstrType_CONSTR_UNIQUE:
			ct := conv.SrcSchema[table]
			ct.Indexes = append(ct.Indexes, schema.Index{Name: c.name, Unique: true, Keys: toSchemaKeys(conv, table, c.cols)})
			conv.SrcSchema[table] = ct
		default:
			ct := conv.SrcSchema[table]
			updateCols(c.ct, c.cols, ct.ColDefs)
			conv.SrcSchema[table] = ct
		}
	}
}
func updateCols(ct pg_query.ConstrType, colNames []string, colDef map[string]schema.Column) {
	for _, c := range colNames {
		cd := colDef[c]
		switch ct {
		case pg_query.ConstrType_CONSTR_NOTNULL:
			cd.NotNull = true
		case pg_query.ConstrType_CONSTR_DEFAULT:
			cd.Ignored.Default = true
		}
		colDef[c] = cd
	}
}