func()

in sources/common/toddl.go [320:423]
93 lines of code
17 McCabe index (conditional complexity)

func (ss *SchemaToSpannerImpl) SchemaToSpannerDDLHelper(conv *internal.Conv, toddl ToDdl, srcTable schema.Table, isRestore bool) error {
	spTableName, err := internal.GetSpannerTable(conv, srcTable.Id)
	if err != nil {
		conv.Unexpected(fmt.Sprintf("Couldn't map source table %s to Spanner: %s", srcTable.Name, err))
		return err
	}
	var spColIds []string
	spColDef := make(map[string]ddl.ColumnDef)

	var (
		totalNonKeyColumnSize int
		tableLevelIssues      []internal.SchemaIssue
	)

	columnLevelIssues := make(map[string][]internal.SchemaIssue)

	// Iterate over columns using ColNames order.
	for _, srcColId := range srcTable.ColIds {
		srcCol := srcTable.ColDefs[srcColId]
		colName, err := internal.GetSpannerCol(conv, srcTable.Id, srcCol.Id, spColDef)
		if err != nil {
			conv.Unexpected(fmt.Sprintf("Couldn't map source column %s of table %s to Spanner: %s", srcTable.Name, srcCol.Name, err))
			continue
		}
		spColIds = append(spColIds, srcColId)
		isPk := IsPrimaryKey(srcColId, srcTable)
		ty, issues := toddl.ToSpannerType(conv, "", srcCol.Type, isPk)

		// TODO(hengfeng): add issues for all elements of srcCol.Ignored.
		if srcCol.Ignored.ForeignKey {
			issues = append(issues, internal.ForeignKey)
		}
		_, isChanged := internal.FixName(srcCol.Name)
		if isChanged && (srcCol.Name != colName) {
			issues = append(issues, internal.IllegalName)
		}
		if srcCol.Ignored.Default {
			issues = append(issues, internal.DefaultValue)
		}
		if srcCol.Ignored.AutoIncrement { // TODO(adibh) - check why this is not there in postgres
			issues = append(issues, internal.AutoIncrement)
		}
		// Set the not null constraint to false for unsupported source datatypes
		isNotNull := srcCol.NotNull
		if findSchemaIssue(issues, internal.NoGoodType) != -1 {
			isNotNull = false
		}
		// Set the not null constraint to false for array datatype and add a warning because
		// datastream does not support array datatypes.
		if ty.IsArray {
			issues = append(issues, internal.ArrayTypeNotSupported)
			isNotNull = false
		}
		// Set auto generation for column
		srcAutoGen := srcCol.AutoGen
		var autoGenCol *ddl.AutoGenCol = &ddl.AutoGenCol{}
		if srcAutoGen.Name != "" {
			autoGenCol, err = toddl.GetColumnAutoGen(conv, srcAutoGen, srcColId, srcTable.Id)
			if autoGenCol != nil {
				if err != nil {
					srcCol.Ignored.AutoIncrement = true
					issues = append(issues, internal.AutoIncrement)
				} else {
					issues = append(issues, internal.SequenceCreated)
				}
			}
		}
		if len(issues) > 0 {
			columnLevelIssues[srcColId] = issues
		}

		spColDef[srcColId] = ddl.ColumnDef{
			Name:    colName,
			T:       ty,
			NotNull: isNotNull,
			Comment: "From: " + quoteIfNeeded(srcCol.Name) + " " + srcCol.Type.Print(),
			Id:      srcColId,
			AutoGen: *autoGenCol,
		}
		if !checkIfColumnIsPartOfPK(srcColId, srcTable.PrimaryKeys) {
			totalNonKeyColumnSize += getColumnSize(ty.Name, ty.Len)
		}
	}
	if totalNonKeyColumnSize > ddl.MaxNonKeyColumnLength {
		tableLevelIssues = append(tableLevelIssues, internal.RowLimitExceeded)
	}
	conv.SchemaIssues[srcTable.Id] = internal.TableIssues{
		TableLevelIssues:  tableLevelIssues,
		ColumnLevelIssues: columnLevelIssues,
	}
	comment := "Spanner schema for source table " + quoteIfNeeded(srcTable.Name)
	conv.SpSchema[srcTable.Id] = ddl.CreateTable{
		Name:             spTableName,
		ColIds:           spColIds,
		ColDefs:          spColDef,
		PrimaryKeys:      cvtPrimaryKeys(srcTable.PrimaryKeys),
		ForeignKeys:      cvtForeignKeys(conv, spTableName, srcTable.Id, srcTable.ForeignKeys, isRestore),
		CheckConstraints: cvtCheckConstraint(conv, srcTable.CheckConstraints),
		Indexes:          cvtIndexes(conv, srcTable.Id, srcTable.Indexes, spColIds, spColDef),
		Comment:          comment,
		Id:               srcTable.Id,
	}
	return nil
}