Store status on activate task

No description

---

Pull Request resolved: #126
commit_hash:aa86b1b4e203823526f65e5b6d87cccfdba82864

cmd/trcli/activate/activate.go

@@ -60,6 +60,9 @@ func RunActivate(
 			time.Sleep(extraWait)
 		}
 	}()
+	if err := cp.RemoveTransferState(transfer.ID, []string{"status"}); err != nil {
+		return xerrors.Errorf("unable to cleanup status state: %w", err)
+	}
 	logger.Log.Infof("run activate with: %T", cp)
 	op := new(model.TransferOperation)
 	op.OperationID = transfer.ID + "/activation"
@@ -87,5 +90,17 @@ func RunActivate(
 	for _, p := range pcp.Progress() {
 		logger.Log.Infof("	part: %s 👌 %v rows in %v", p.String(), p.CompletedRows, time.Since(st))
 	}
+	if err := cp.SetTransferState(transfer.ID, map[string]*coordinator.TransferStateData{
+		"status": {
+			Generic:             "activated",
+			IncrementalTables:   nil,
+			OraclePosition:      nil,
+			MysqlGtid:           nil,
+			MysqlBinlogPosition: nil,
+			YtStaticPart:        nil,
+		},
+	}); err != nil {
+		return xerrors.Errorf("unable to set transfer state: %w", err)
+	}
 	return nil
 }

pkg/providers/clickhouse/sink_cluster.go

@@ -103,7 +103,7 @@ func (c *sinkCluster) DropTable(tableName string) error {
 func (c *sinkCluster) execDDL(executor func(distributed bool) error) error {
 	c.distributedDDLMu.Lock()
 
-	if c.distributedDDLEnabled == nil && c.topology.ClusterName() == "" {
+	if c.distributedDDLEnabled == nil && (c.topology.ClusterName() == "" || c.topology.SingleNode()) {
 		if !c.topology.SingleNode() {
 			return xerrors.Errorf("resolved empty cluster name for non-single-node cluster")
 		}