Merge branch 'master' into named-panic

Shlomi Noach · web-flow · commit ff17d2f844d0 · 2019-02-25T14:00:12.000+02:00
diff --git a/RELEASE_VERSION b/RELEASE_VERSION
@@ -1 +1 @@
-1.0.47
+1.0.48
diff --git a/build.sh b/build.sh
@@ -42,8 +42,8 @@ function build {
     builddir=$(setuptree)
     cp $buildpath/$target $builddir/gh-ost/usr/bin
     cd $buildpath
-    fpm -v "${RELEASE_VERSION}" --epoch 1 -f -s dir -n gh-ost -m shlomi-noach --description "GitHub's Online Schema Migrations for MySQL " --url "https://github.com/github/gh-ost" --vendor "GitHub" --license "Apache 2.0" -C $builddir/gh-ost --prefix=/ -t rpm .
-    fpm -v "${RELEASE_VERSION}" --epoch 1 -f -s dir -n gh-ost -m shlomi-noach --description "GitHub's Online Schema Migrations for MySQL " --url "https://github.com/github/gh-ost" --vendor "GitHub" --license "Apache 2.0" -C $builddir/gh-ost --prefix=/ -t deb --deb-no-default-config-files .
+    fpm -v "${RELEASE_VERSION}" --epoch 1 -f -s dir -n gh-ost -m 'shlomi-noach <shlomi-noach+gh-ost-deb@github.com>' --description "GitHub's Online Schema Migrations for MySQL " --url "https://github.com/github/gh-ost" --vendor "GitHub" --license "Apache 2.0" -C $builddir/gh-ost --prefix=/ -t rpm .
+    fpm -v "${RELEASE_VERSION}" --epoch 1 -f -s dir -n gh-ost -m 'shlomi-noach <shlomi-noach+gh-ost-deb@github.com>' --description "GitHub's Online Schema Migrations for MySQL " --url "https://github.com/github/gh-ost" --vendor "GitHub" --license "Apache 2.0" -C $builddir/gh-ost --prefix=/ -t deb --deb-no-default-config-files .
   fi
 }
 
diff --git a/doc/command-line-flags.md b/doc/command-line-flags.md
@@ -185,6 +185,18 @@ By default `gh-ost` verifies no foreign keys exist on the migrated table. On ser
 
 See [`approve-renamed-columns`](#approve-renamed-columns)
 
+### ssl
+
+By default `gh-ost` does not use ssl/tls connections to the database servers when performing migrations. This flag instructs `gh-ost` to use encrypted connections. If enabled, `gh-ost` will use the system's ca certificate pool for server certificate verification. If a different certificate is needed for server verification, see `--ssl-ca`. If you wish to skip server verification, but still use encrypted connections, use with `--ssl-allow-insecure`.
+
+### ssl-allow-insecure
+
+Allows `gh-ost` to connect to the MySQL servers using encrypted connections, but without verifying the validity of the certificate provided by the server during the connection. Requires `--ssl`.
+
+### ssl-ca
+
+`--ssl-ca=/path/to/ca-cert.pem`: ca certificate file (in PEM format) to use for server certificate verification. If specified, the default system ca cert pool will not be used for verification, only the ca cert provided here. Requires `--ssl`.
+
 ### test-on-replica
 
 Issue the migration on a replica; do not modify data on master. Useful for validating, testing and benchmarking. See [`testing-on-replica`](testing-on-replica.md)
diff --git a/doc/hooks.md b/doc/hooks.md
@@ -69,6 +69,8 @@ The following variables are available on all hooks:
 - `GH_OST_INSPECTED_HOST`
 - `GH_OST_EXECUTING_HOST`
 - `GH_OST_HOOKS_HINT` - copy of `--hooks-hint` value
+- `GH_OST_HOOKS_HINT_OWNER` - copy of `--hooks-hint-owner` value
+- `GH_OST_HOOKS_HINT_TOKEN` - copy of `--hooks-hint-token` value
 - `GH_OST_DRY_RUN` - whether or not the `gh-ost` run is a dry run
 
 The following variable are available on particular hooks:
diff --git a/doc/questions.md b/doc/questions.md
@@ -28,3 +28,9 @@ It is therefore unlikely that `gh-ost` will support this behavior.
 Yes. TL;DR if running all on same replica/master, make sure to provide `--replica-server-id`. [Read more](cheatsheet.md#concurrent-migrations)
 
 # Why
+
+### Why Is the "Connect to Replica" mode preferred? 
+
+To avoid placing extra load on the master. `gh-ost` connects as a replication client. Each additional replica adds some load to the master. 
+
+To monitor replication lag from a replica. This makes the replication lag throttle, `--max-lag-millis`, more representative of the lag experienced by other replicas following the master (perhaps N levels deep in a tree of replicas).
diff --git a/go/base/context.go b/go/base/context.go
@@ -99,6 +99,9 @@ type MigrationContext struct {
 	ConfigFile        string
 	CliUser           string
 	CliPassword       string
+	UseTLS            bool
+	TLSAllowInsecure  bool
+	TLSCACertificate  string
 	CliMasterUser     string
 	CliMasterPassword string
 
@@ -127,6 +130,8 @@ type MigrationContext struct {
 	PanicFlagFile                       string
 	HooksPath                           string
 	HooksHintMessage                    string
+	HooksHintOwner                      string
+	HooksHintToken                      string
 
 	DropServeSocket bool
 	ServeSocketFile string
@@ -696,6 +701,13 @@ func (this *MigrationContext) ApplyCredentials() {
 	}
 }
 
+func (this *MigrationContext) SetupTLS() error {
+	if this.UseTLS {
+		return this.InspectorConnectionConfig.UseTLS(this.TLSCACertificate, this.TLSAllowInsecure)
+	}
+	return nil
+}
+
 // ReadConfigFile attempts to read the config file, if it exists
 func (this *MigrationContext) ReadConfigFile() error {
 	this.configMutex.Lock()
diff --git a/go/binlog/binlog_entry.go b/go/binlog/binlog_entry.go
@@ -26,7 +26,7 @@ func NewBinlogEntry(logFile string, logPos uint64) *BinlogEntry {
 	return binlogEntry
 }
 
-// NewBinlogEntry creates an empty, ready to go BinlogEntry object
+// NewBinlogEntryAt creates an empty, ready to go BinlogEntry object
 func NewBinlogEntryAt(coordinates mysql.BinlogCoordinates) *BinlogEntry {
 	binlogEntry := &BinlogEntry{
 		Coordinates: coordinates,
@@ -41,7 +41,7 @@ func (this *BinlogEntry) Duplicate() *BinlogEntry {
 	return binlogEntry
 }
 
-// Duplicate creates and returns a new binlog entry, with some of the attributes pre-assigned
+// String() returns a string representation of this binlog entry
 func (this *BinlogEntry) String() string {
 	return fmt.Sprintf("[BinlogEntry at %+v; dml:%+v]", this.Coordinates, this.DmlEvent)
 }
diff --git a/go/binlog/gomysql_reader.go b/go/binlog/gomysql_reader.go
@@ -46,6 +46,7 @@ func NewGoMySQLReader(migrationContext *base.MigrationContext) (binlogReader *Go
 		Port:       uint16(binlogReader.connectionConfig.Key.Port),
 		User:       binlogReader.connectionConfig.User,
 		Password:   binlogReader.connectionConfig.Password,
+		TLSConfig:  binlogReader.connectionConfig.TLSConfig(),
 		UseDecimal: true,
 	}
 	binlogReader.binlogSyncer = replication.NewBinlogSyncer(binlogSyncerConfig)
@@ -112,7 +113,7 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
 			}
 		}
-		// The channel will do the throttling. Whoever is reding from the channel
+		// The channel will do the throttling. Whoever is reading from the channel
 		// decides whether action is taken synchronously (meaning we wait before
 		// next iteration) or asynchronously (we keep pushing more events)
 		// In reality, reads will be synchronous
diff --git a/go/cmd/gh-ost/main.go b/go/cmd/gh-ost/main.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/logic"
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/outbrain/golib/log"
 
 	"golang.org/x/crypto/ssh/terminal"
@@ -54,6 +55,10 @@ func main() {
 	flag.StringVar(&migrationContext.ConfigFile, "conf", "", "Config file")
 	askPass := flag.Bool("ask-pass", false, "prompt for MySQL password")
 
+	flag.BoolVar(&migrationContext.UseTLS, "ssl", false, "Enable SSL encrypted connections to MySQL hosts")
+	flag.StringVar(&migrationContext.TLSCACertificate, "ssl-ca", "", "CA certificate in PEM format for TLS connections to MySQL hosts. Requires --ssl")
+	flag.BoolVar(&migrationContext.TLSAllowInsecure, "ssl-allow-insecure", false, "Skips verification of MySQL hosts' certificate chain and host name. Requires --ssl")
+
 	flag.StringVar(&migrationContext.DatabaseName, "database", "", "database name (mandatory)")
 	flag.StringVar(&migrationContext.OriginalTableName, "table", "", "table name (mandatory)")
 	flag.StringVar(&migrationContext.AlterStatement, "alter", "", "alter statement (mandatory)")
@@ -110,6 +115,8 @@ func main() {
 
 	flag.StringVar(&migrationContext.HooksPath, "hooks-path", "", "directory where hook files are found (default: empty, ie. hooks disabled). Hook files found on this path, and conforming to hook naming conventions will be executed")
 	flag.StringVar(&migrationContext.HooksHintMessage, "hooks-hint", "", "arbitrary message to be injected to hooks via GH_OST_HOOKS_HINT, for your convenience")
+	flag.StringVar(&migrationContext.HooksHintOwner, "hooks-hint-owner", "", "arbitrary name of owner to be injected to hooks via GH_OST_HOOKS_HINT_OWNER, for your convenience")
+	flag.StringVar(&migrationContext.HooksHintToken, "hooks-hint-token", "", "arbitrary token to be injected to hooks via GH_OST_HOOKS_HINT_TOKEN, for your convenience")
 
 	flag.UintVar(&migrationContext.ReplicaServerId, "replica-server-id", 99999, "server id used by gh-ost process. Default: 99999")
 
@@ -195,6 +202,12 @@ func main() {
 	if migrationContext.CliMasterPassword != "" && migrationContext.AssumeMasterHostname == "" {
 		log.Fatalf("--master-password requires --assume-master-host")
 	}
+	if migrationContext.TLSCACertificate != "" && !migrationContext.UseTLS {
+		log.Fatalf("--ssl-ca requires --ssl")
+	}
+	if migrationContext.TLSAllowInsecure && !migrationContext.UseTLS {
+		log.Fatalf("--ssl-allow-insecure requires --ssl")
+	}
 	if *replicationLagQuery != "" {
 		log.Warningf("--replication-lag-query is deprecated")
 	}
@@ -239,6 +252,9 @@ func main() {
 	migrationContext.SetThrottleHTTP(*throttleHTTP)
 	migrationContext.SetDefaultNumRetries(*defaultRetries)
 	migrationContext.ApplyCredentials()
+	if err := migrationContext.SetupTLS(); err != nil {
+		log.Fatale(err)
+	}
 	if err := migrationContext.SetCutOverLockTimeoutSeconds(*cutOverLockTimeoutSeconds); err != nil {
 		log.Errore(err)
 	}
diff --git a/go/logic/applier.go b/go/logic/applier.go
@@ -73,7 +73,7 @@ func (this *Applier) InitDBConnections() (err error) {
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, applierUri); err != nil {
 		return err
 	}
-	singletonApplierUri := fmt.Sprintf("%s?timeout=0", applierUri)
+	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if this.singletonDB, _, err = mysql.GetDB(this.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err
 	}
@@ -126,7 +126,6 @@ func (this *Applier) readTableColumns() (err error) {
 
 // showTableStatus returns the output of `show table status like '...'` command
 func (this *Applier) showTableStatus(tableName string) (rowMap sqlutils.RowMap) {
-	rowMap = nil
 	query := fmt.Sprintf(`show /* gh-ost */ table status from %s like '%s'`, sql.EscapeName(this.migrationContext.DatabaseName), tableName)
 	sqlutils.QueryRowsMap(this.db, query, func(m sqlutils.RowMap) error {
 		rowMap = m
@@ -482,6 +481,7 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
 		if err != nil {
 			return nil, err
 		}
+		defer tx.Rollback()
 		sessionQuery := fmt.Sprintf(`SET
 			SESSION time_zone = '%s',
 			sql_mode = CONCAT(@@session.sql_mode, ',STRICT_ALL_TABLES')
@@ -1001,15 +1001,19 @@ func (this *Applier) ApplyDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) error {
 			if err != nil {
 				return err
 			}
+			rollback := func(err error) error {
+				tx.Rollback()
+				return err
+			}
 			sessionQuery := `SET
 			SESSION time_zone = '+00:00',
 			sql_mode = CONCAT(@@session.sql_mode, ',STRICT_ALL_TABLES')
 			`
 			if _, err := tx.Exec(sessionQuery); err != nil {
-				return err
+				return rollback(err)
 			}
 			if _, err := tx.Exec(buildResult.query, buildResult.args...); err != nil {
-				return err
+				return rollback(err)
 			}
 			if err := tx.Commit(); err != nil {
 				return err
diff --git a/go/logic/hooks.go b/go/logic/hooks.go
@@ -64,6 +64,8 @@ func (this *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) [
 	env = append(env, fmt.Sprintf("GH_OST_INSPECTED_HOST=%s", this.migrationContext.GetInspectorHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_EXECUTING_HOST=%s", this.migrationContext.Hostname))
 	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT=%s", this.migrationContext.HooksHintMessage))
+	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_OWNER=%s", this.migrationContext.HooksHintOwner))
+	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_TOKEN=%s", this.migrationContext.HooksHintToken))
 	env = append(env, fmt.Sprintf("GH_OST_DRY_RUN=%t", this.migrationContext.Noop))
 
 	for _, variable := range extraVariables {
diff --git a/go/logic/inspect.go b/go/logic/inspect.go
@@ -622,8 +622,6 @@ func (this *Inspector) getCandidateUniqueKeys(tableName string) (uniqueKeys [](*
       GROUP BY TABLE_SCHEMA, TABLE_NAME, INDEX_NAME
     ) AS UNIQUES
     ON (
-      COLUMNS.TABLE_SCHEMA = UNIQUES.TABLE_SCHEMA AND
-      COLUMNS.TABLE_NAME = UNIQUES.TABLE_NAME AND
       COLUMNS.COLUMN_NAME = UNIQUES.FIRST_COLUMN_NAME
     )
     WHERE
@@ -692,14 +690,17 @@ func (this *Inspector) getSharedColumns(originalColumns, ghostColumns *sql.Colum
 		for _, ghostColumn := range ghostColumns.Names() {
 			if strings.EqualFold(originalColumn, ghostColumn) {
 				isSharedColumn = true
+				break
 			}
 			if strings.EqualFold(columnRenameMap[originalColumn], ghostColumn) {
 				isSharedColumn = true
+				break
 			}
 		}
 		for droppedColumn := range this.migrationContext.DroppedColumnsMap {
 			if strings.EqualFold(originalColumn, droppedColumn) {
 				isSharedColumn = false
+				break
 			}
 		}
 		for _, virtualColumn := range originalVirtualColumns.Names() {
@@ -758,9 +759,8 @@ func (this *Inspector) getMasterConnectionConfig() (applierConfig *mysql.Connect
 }
 
 func (this *Inspector) getReplicationLag() (replicationLag time.Duration, err error) {
-	replicationLag, err = mysql.GetReplicationLag(
+	replicationLag, err = mysql.GetReplicationLagFromSlaveStatus(
 		this.informationSchemaDb,
-		this.migrationContext.InspectorConnectionConfig,
 	)
 	return replicationLag, err
 }
diff --git a/go/logic/migrator.go b/go/logic/migrator.go
@@ -78,7 +78,7 @@ type Migrator struct {
 
 	rowCopyCompleteFlag int64
 	// copyRowsQueue should not be buffered; if buffered some non-damaging but
-	//  excessive work happens at the end of the iteration as new copy-jobs arrive befroe realizing the copy is complete
+	//  excessive work happens at the end of the iteration as new copy-jobs arrive before realizing the copy is complete
 	copyRowsQueue    chan tableWriteFunc
 	applyEventsQueue chan *applyEventStruct
 
diff --git a/go/logic/throttler.go b/go/logic/throttler.go
@@ -140,8 +140,8 @@ func (this *Throttler) collectReplicationLag(firstThrottlingCollected chan<- boo
 		if this.migrationContext.TestOnReplica || this.migrationContext.MigrateOnReplica {
 			// when running on replica, the heartbeat injection is also done on the replica.
 			// This means we will always get a good heartbeat value.
-			// When runnign on replica, we should instead check the `SHOW SLAVE STATUS` output.
-			if lag, err := mysql.GetReplicationLag(this.inspector.informationSchemaDb, this.inspector.connectionConfig); err != nil {
+			// When running on replica, we should instead check the `SHOW SLAVE STATUS` output.
+			if lag, err := mysql.GetReplicationLagFromSlaveStatus(this.inspector.informationSchemaDb); err != nil {
 				return log.Errore(err)
 			} else {
 				atomic.StoreInt64(&this.migrationContext.CurrentLag, int64(lag))
diff --git a/go/mysql/connection.go b/go/mysql/connection.go
@@ -6,8 +6,14 @@
 package mysql
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
 	"fmt"
+	"io/ioutil"
 	"net"
+
+	"github.com/go-sql-driver/mysql"
 )
 
 // ConnectionConfig is the minimal configuration required to connect to a MySQL server
@@ -16,6 +22,7 @@ type ConnectionConfig struct {
 	User       string
 	Password   string
 	ImpliedKey *InstanceKey
+	tlsConfig  *tls.Config
 }
 
 func NewConnectionConfig() *ConnectionConfig {
@@ -29,9 +36,10 @@ func NewConnectionConfig() *ConnectionConfig {
 // DuplicateCredentials creates a new connection config with given key and with same credentials as this config
 func (this *ConnectionConfig) DuplicateCredentials(key InstanceKey) *ConnectionConfig {
 	config := &ConnectionConfig{
-		Key:      key,
-		User:     this.User,
-		Password: this.Password,
+		Key:       key,
+		User:      this.User,
+		Password:  this.Password,
+		tlsConfig: this.tlsConfig,
 	}
 	config.ImpliedKey = &config.Key
 	return config
@@ -42,13 +50,47 @@ func (this *ConnectionConfig) Duplicate() *ConnectionConfig {
 }
 
 func (this *ConnectionConfig) String() string {
-	return fmt.Sprintf("%s, user=%s", this.Key.DisplayString(), this.User)
+	return fmt.Sprintf("%s, user=%s, usingTLS=%t", this.Key.DisplayString(), this.User, this.tlsConfig != nil)
 }
 
 func (this *ConnectionConfig) Equals(other *ConnectionConfig) bool {
 	return this.Key.Equals(&other.Key) || this.ImpliedKey.Equals(other.ImpliedKey)
 }
 
+func (this *ConnectionConfig) UseTLS(caCertificatePath string, allowInsecure bool) error {
+	var rootCertPool *x509.CertPool
+	var err error
+
+	if !allowInsecure {
+		if caCertificatePath == "" {
+			rootCertPool, err = x509.SystemCertPool()
+			if err != nil {
+				return err
+			}
+		} else {
+			rootCertPool = x509.NewCertPool()
+			pem, err := ioutil.ReadFile(caCertificatePath)
+			if err != nil {
+				return err
+			}
+			if ok := rootCertPool.AppendCertsFromPEM(pem); !ok {
+				return errors.New("could not add ca certificate to cert pool")
+			}
+		}
+	}
+
+	this.tlsConfig = &tls.Config{
+		RootCAs:            rootCertPool,
+		InsecureSkipVerify: allowInsecure,
+	}
+
+	return mysql.RegisterTLSConfig(this.Key.StringCode(), this.tlsConfig)
+}
+
+func (this *ConnectionConfig) TLSConfig() *tls.Config {
+	return this.tlsConfig
+}
+
 func (this *ConnectionConfig) GetDBUri(databaseName string) string {
 	hostname := this.Key.Hostname
 	var ip = net.ParseIP(hostname)
@@ -57,5 +99,11 @@ func (this *ConnectionConfig) GetDBUri(databaseName string) string {
 		hostname = fmt.Sprintf("[%s]", hostname)
 	}
 	interpolateParams := true
-	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?interpolateParams=%t&autocommit=true&charset=utf8mb4,utf8,latin1", this.User, this.Password, hostname, this.Key.Port, databaseName, interpolateParams)
+	// go-mysql-driver defaults to false if tls param is not provided; explicitly setting here to
+	// simplify construction of the DSN below.
+	tlsOption := "false"
+	if this.tlsConfig != nil {
+		tlsOption = this.Key.StringCode()
+	}
+	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?interpolateParams=%t&autocommit=true&charset=utf8mb4,utf8,latin1&tls=%s", this.User, this.Password, hostname, this.Key.Port, databaseName, interpolateParams, tlsOption)
 }
diff --git a/go/mysql/connection_test.go b/go/mysql/connection_test.go
diff --git a/go/mysql/utils.go b/go/mysql/utils.go
diff --git a/go/sql/builder.go b/go/sql/builder.go
diff --git a/localtests/datetime-submillis/create.sql b/localtests/datetime-submillis/create.sql
diff --git a/localtests/test.sh b/localtests/test.sh

Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ func NewBinlogEntry(logFile string, logPos uint64) *BinlogEntry {`
`26`	`26`	`return binlogEntry`
`27`	`27`	`}`
`28`	`28`
`29`		`-// NewBinlogEntry creates an empty, ready to go BinlogEntry object`
	`29`	`+// NewBinlogEntryAt creates an empty, ready to go BinlogEntry object`
`30`	`30`	`func NewBinlogEntryAt(coordinates mysql.BinlogCoordinates) *BinlogEntry {`
`31`	`31`	`binlogEntry := &BinlogEntry{`
`32`	`32`	`Coordinates: coordinates,`
`@@ -41,7 +41,7 @@ func (this BinlogEntry) Duplicate() BinlogEntry {`
`41`	`41`	`return binlogEntry`
`42`	`42`	`}`
`43`	`43`
`44`		`-// Duplicate creates and returns a new binlog entry, with some of the attributes pre-assigned`
	`44`	`+// String() returns a string representation of this binlog entry`
`45`	`45`	`func (this *BinlogEntry) String() string {`
`46`	`46`	`return fmt.Sprintf("[BinlogEntry at %+v; dml:%+v]", this.Coordinates, this.DmlEvent)`
`47`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -46,6 +46,7 @@ func NewGoMySQLReader(migrationContext base.MigrationContext) (binlogReader Go`
`46`	`46`	`Port: uint16(binlogReader.connectionConfig.Key.Port),`
`47`	`47`	`User: binlogReader.connectionConfig.User,`
`48`	`48`	`Password: binlogReader.connectionConfig.Password,`
	`49`	`+ TLSConfig: binlogReader.connectionConfig.TLSConfig(),`
`49`	`50`	`UseDecimal: true,`
`50`	`51`	`}`
`51`	`52`	`binlogReader.binlogSyncer = replication.NewBinlogSyncer(binlogSyncerConfig)`
`@@ -112,7 +113,7 @@ func (this GoMySQLReader) handleRowsEvent(ev replication.BinlogEvent, rowsEven`
`112`	`113`	`binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)`
`113`	`114`	`}`
`114`	`115`	`}`
`115`		`- // The channel will do the throttling. Whoever is reding from the channel`
	`116`	`+ // The channel will do the throttling. Whoever is reading from the channel`
`116`	`117`	`// decides whether action is taken synchronously (meaning we wait before`
`117`	`118`	`// next iteration) or asynchronously (we keep pushing more events)`
`118`	`119`	`// In reality, reads will be synchronous`