[wip] attempt to get into a working state

Author: kamushadenes

cmd/trcli/main.go

@@ -1,11 +1,9 @@
 package main
 
 import (
-	"context"
 	"net/http"
 	"os"
 	"strings"
-	"time"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
@@ -21,7 +19,6 @@ import (
 	"github.com/transferia/transferia/pkg/abstract"
 	coordinator "github.com/transferia/transferia/pkg/abstract/coordinator"
 	"github.com/transferia/transferia/pkg/cobraaux"
-	"github.com/transferia/transferia/pkg/coordinator/etcdcoordinator"
 	"github.com/transferia/transferia/pkg/coordinator/s3coordinator"
 	_ "github.com/transferia/transferia/pkg/dataplane"
 	"github.com/transferia/transferia/pkg/serverutil"
@@ -48,12 +45,6 @@ func main() {
 	logConfig := defaultLogConfig
 	coordinatorTyp := defaultCoordinator
 	coordinatorS3Bucket := ""
-	coordinatorEtcdEndpoints := []string{}
-	coordinatorEtcdUsername := ""
-	coordinatorEtcdPassword := ""
-	coordinatorEtcdCertFile := ""
-	coordinatorEtcdKeyFile := ""
-	coordinatorEtcdCAFile := ""
 	runProfiler := false
 
 	promRegistry, registry := internal_metrics.NewPrometheusRegistryWithNameProcessor()
@@ -63,11 +54,7 @@ func main() {
 		Short:        "Transferia cli",
 		Example:      "./trcli help",
 		SilenceUsage: true,
-		Version:      getVersionString(),
 		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
-			ctx := context.Background()
-			cmd.SetContext(ctx)
-
 			if strings.Contains(cmd.CommandPath(), "describe") {
 				return nil
 			}
@@ -81,14 +68,7 @@ func main() {
 					ErrorHandling: promhttp.PanicOnError,
 				}))
 				logger.Log.Infof("Prometheus is uprising on port %v", "9091")
-				server := &http.Server{
-					Addr:         ":9091",
-					ReadTimeout:  60 * time.Second,
-					WriteTimeout: 60 * time.Second,
-					Handler:      rootMux,
-				}
-
-				if err := server.ListenAndServe(); err != nil {
+				if err := http.ListenAndServe(":9091", rootMux); err != nil {
 					logger.Log.Error("failed to serve metrics", log.Error(err))
 				}
 			}()
@@ -137,19 +117,6 @@ func main() {
 				if rt.CurrentJob > 0 || rt.ShardingUpload.JobCount > 1 {
 					return xerrors.Errorf("for sharding upload memory coordinator won't work")
 				}
-			case "etcd":
-				var err error
-				cp, err = etcdcoordinator.NewEtcdCoordinator(cmd.Context(), etcdcoordinator.EtcdConfig{
-					Endpoints: coordinatorEtcdEndpoints,
-					Username:  coordinatorEtcdUsername,
-					Password:  coordinatorEtcdPassword,
-					CertFile:  coordinatorEtcdCertFile,
-					KeyFile:   coordinatorEtcdKeyFile,
-					CAFile:    coordinatorEtcdCAFile,
-				}, logger.Log)
-				if err != nil {
-					return xerrors.Errorf("unable to load etcd coordinator: %w", err)
-				}
 			case "s3":
 				var err error
 				cp, err = s3coordinator.NewS3(coordinatorS3Bucket, logger.Log)
@@ -172,16 +139,8 @@ func main() {
 
 	rootCommand.PersistentFlags().StringVar(&logLevel, "log-level", defaultLogLevel, "Specifies logging level for output logs (\"panic\", \"fatal\", \"error\", \"warning\", \"info\", \"debug\")")
 	rootCommand.PersistentFlags().StringVar(&logConfig, "log-config", defaultLogConfig, "Specifies logging config for output logs (\"console\", \"json\", \"minimal\")")
-
-	rootCommand.PersistentFlags().StringVar(&coordinatorTyp, "coordinator", defaultCoordinator, "Specifies how to coordinate transfer nodes (\"memory\", \"s3\", \"etcd\")")
+	rootCommand.PersistentFlags().StringVar(&coordinatorTyp, "coordinator", defaultCoordinator, "Specifies how to coordinate transfer nodes (\"memory\", \"s3\")")
 	rootCommand.PersistentFlags().StringVar(&coordinatorS3Bucket, "coordinator-s3-bucket", "", "Bucket for s3 coordinator")
-	rootCommand.PersistentFlags().StringSliceVar(&coordinatorEtcdEndpoints, "coordinator-etcd-endpoints", []string{"http://localhost:2379"}, "Endpoints for etcd coordinator")
-	rootCommand.PersistentFlags().StringVar(&coordinatorEtcdUsername, "coordinator-etcd-username", "", "Username for etcd coordinator")
-	rootCommand.PersistentFlags().StringVar(&coordinatorEtcdPassword, "coordinator-etcd-password", "", "Password for etcd coordinator")
-	rootCommand.PersistentFlags().StringVar(&coordinatorEtcdCertFile, "coordinator-etcd-cert-file", "", "Path to the etcd client certificate file")
-	rootCommand.PersistentFlags().StringVar(&coordinatorEtcdKeyFile, "coordinator-etcd-key-file", "", "Path to the etcd client key file")
-	rootCommand.PersistentFlags().StringVar(&coordinatorEtcdCAFile, "coordinator-etcd-ca-file", "", "Path to the etcd CA certificate file")
-
 	rootCommand.PersistentFlags().BoolVar(&runProfiler, "run-profiler", true, "Run go pprof for performance profiles on 8080 port")
 	rootCommand.PersistentFlags().IntVar(&rt.CurrentJob, "coordinator-job-index", 0, "Worker job index")
 	rootCommand.PersistentFlags().IntVar(&rt.ShardingUpload.JobCount, "coordinator-job-count", 0, "Worker job count, if more then 1 - run consider as sharded, coordinator is required to be non memory")

go.mod

@@ -151,8 +151,6 @@ require (
 	github.com/containerd/containerd v1.7.25 // indirect
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/containerd/platforms v0.2.1 // indirect
-	github.com/coreos/go-semver v0.3.0 // indirect
-	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/creack/pty v1.1.21 // indirect
 	github.com/cyphar/filepath-securejoin v0.4.1 // indirect
@@ -190,7 +188,7 @@ require (
 	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
 	github.com/gorilla/css v1.0.1 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
-	github.com/imdario/mergo v0.3.15 // indirect
+	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
@@ -265,8 +263,6 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	github.com/zeebo/assert v1.3.1 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
-	go.etcd.io/etcd/api/v3 v3.5.21 // indirect
-	go.etcd.io/etcd/client/pkg/v3 v3.5.21 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.36.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect

go.sum

@@ -862,8 +862,8 @@ github.com/Microsoft/hcsshim v0.9.4/go.mod h1:7pLA8lDk46WKDWlVsENo92gC0XFa8rbKfy
 github.com/Microsoft/hcsshim v0.9.6/go.mod h1:7pLA8lDk46WKDWlVsENo92gC0XFa8rbKfyFRBqxEbCc=
 github.com/Microsoft/hcsshim v0.9.10/go.mod h1:7pLA8lDk46WKDWlVsENo92gC0XFa8rbKfyFRBqxEbCc=
 github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
-github.com/Microsoft/hcsshim v0.13.0 h1:/BcXOiS6Qi7N9XqUcv27vkIuVOkBEcWstd2pMlWSeaA=
-github.com/Microsoft/hcsshim v0.13.0/go.mod h1:9KWJ/8DgU+QzYGupX4tzMhRQE8h6w90lH6HAaclpEok=
+github.com/Microsoft/hcsshim v0.11.7 h1:vl/nj3Bar/CvJSYo7gIQPyRWc9f3c6IeSNavBTSZNZQ=
+github.com/Microsoft/hcsshim v0.11.7/go.mod h1:MV8xMfmECjl5HdO7U/3/hFVnkmSBjAjmA09d4bExKcU=
 github.com/Microsoft/hcsshim/test v0.0.0-20201218223536-d3e5debf77da/go.mod h1:5hlzMzRKMLyo42nCZ9oml8AdTlq/0cvIaBv6tK1RehU=
 github.com/Microsoft/hcsshim/test v0.0.0-20210227013316-43a75bb4edd3/go.mod h1:mw7qgWloBUl75W/gVH3cQszUg1+gUITj7D6NY7ywVnY=
 github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ=
@@ -1197,15 +1197,13 @@ github.com/coreos/go-iptables v0.5.0/go.mod h1:/mVI274lEDI2ns62jHCDnCyBF9Iwsmeka
 github.com/coreos/go-iptables v0.6.0/go.mod h1:Qe8Bv2Xik5FyTXwgIbLAnv2sWSBmvWdFETJConOQ//Q=
 github.com/coreos/go-oidc v2.1.0+incompatible/go.mod h1:CgnwVTmzoESiwO9qyAFEMiHoZ1nMCKZlZ9V6mm3/LKc=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
-github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/coreos/go-systemd v0.0.0-20161114122254-48702e0da86b/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd/v22 v22.0.0/go.mod h1:xO0FLkIi5MaZafQlIrOotqXZ90ih+1atmu1JpKERPPk=
 github.com/coreos/go-systemd/v22 v22.1.0/go.mod h1:xO0FLkIi5MaZafQlIrOotqXZ90ih+1atmu1JpKERPPk=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
-github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E=
@@ -1672,9 +1670,8 @@ github.com/imdario/mergo v0.3.6/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJ
 github.com/imdario/mergo v0.3.8/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.10/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
+github.com/imdario/mergo v0.3.12 h1:b6R2BslTbIEToALKP7LxUvijTsNI9TAe80pLWN2g/HU=
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
-github.com/imdario/mergo v0.3.15 h1:M8XP7IuFNsqUx6VPK2P9OSmsYsI/YFaGil0uD21V3dM=
-github.com/imdario/mergo v0.3.15/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.0.1/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
@@ -2354,15 +2351,9 @@ go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/bbolt v1.3.6/go.mod h1:qXsaaIqmgQH0T+OPdb99Bf+PKfBBQVAdyD6TY9G8XM4=
 go.etcd.io/bbolt v1.3.7/go.mod h1:N9Mkw9X8x5fupy0IKsmuqVtoGDyxsaDlbk4Rd05IAQw=
 go.etcd.io/etcd/api/v3 v3.5.5/go.mod h1:KFtNaxGDw4Yx/BA4iPPwevUTAuqcsPxzyX8PHydchN8=
-go.etcd.io/etcd/api/v3 v3.5.21 h1:A6O2/JDb3tvHhiIz3xf9nJ7REHvtEFJJ3veW3FbCnS8=
-go.etcd.io/etcd/api/v3 v3.5.21/go.mod h1:c3aH5wcvXv/9dqIw2Y810LDXJfhSYdHQ0vxmP3CCHVY=
 go.etcd.io/etcd/client/pkg/v3 v3.5.5/go.mod h1:ggrwbk069qxpKPq8/FKkQ3Xq9y39kbFR4LnKszpRXeQ=
-go.etcd.io/etcd/client/pkg/v3 v3.5.21 h1:lPBu71Y7osQmzlflM9OfeIV2JlmpBjqBNlLtcoBqUTc=
-go.etcd.io/etcd/client/pkg/v3 v3.5.21/go.mod h1:BgqT/IXPjK9NkeSDjbzwsHySX3yIle2+ndz28nVsjUs=
 go.etcd.io/etcd/client/v2 v2.305.5/go.mod h1:zQjKllfqfBVyVStbt4FaosoX2iYd8fV/GRy/PbowgP4=
 go.etcd.io/etcd/client/v3 v3.5.5/go.mod h1:aApjR4WGlSumpnJ2kloS75h6aHUmAyaPLjHMxpc7E7c=
-go.etcd.io/etcd/client/v3 v3.5.21 h1:T6b1Ow6fNjOLOtM0xSoKNQt1ASPCLWrF9XMHcH9pEyY=
-go.etcd.io/etcd/client/v3 v3.5.21/go.mod h1:mFYy67IOqmbRf/kRUvsHixzo3iG+1OF2W2+jVIQRAnU=
 go.etcd.io/etcd/pkg/v3 v3.5.5/go.mod h1:6ksYFxttiUGzC2uxyqiyOEvhAiD0tuIqSZkX3TyPdaE=
 go.etcd.io/etcd/raft/v3 v3.5.5/go.mod h1:76TA48q03g1y1VpTue92jZLr9lIHKUNcYdZOOGyx8rI=
 go.etcd.io/etcd/server/v3 v3.5.5/go.mod h1:rZ95vDw/jrvsbj9XpTqPrTAB9/kzchVdhRirySPkUBc=
@@ -2395,8 +2386,8 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.32.0/go.mod h1:
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.35.0/go.mod h1:9NiG9I2aHTKkcxqCILhjtyNA1QEiCjdBACv4IvrFQ+c=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.45.0/go.mod h1:62CPTSry9QZtOaSsE3tOzhx6LzDhHnXJ6xHeMNNiM6Q=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0/go.mod h1:p8pYQP+m5XfbZm9fxtSKAbM6oIllS7s2AfxrChvc7iw=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus=
-go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ=
 go.opentelemetry.io/otel v1.0.1/go.mod h1:OPEOD4jIT2SlZPMmwT6FqZz2C0ZNdQqiWcoK6M0SNFU=
 go.opentelemetry.io/otel v1.3.0/go.mod h1:PWIKzi6JCp7sM0k9yZ43VX+T345uNbAkDKwHVjb2PTs=
 go.opentelemetry.io/otel v1.7.0/go.mod h1:5BdUoMIz5WEs0vt0CUEMtSSaTSHBBVwrhnz7+nrD5xk=

pkg/abstract/async_sink.go

@@ -19,7 +19,7 @@ type AsyncSink interface {
 // AsyncPushConcurrencyErr indicates a Push has been called on an already closed AsyncSink. This must not happen and means there are concurrency issues in the implementation of a source.
 var AsyncPushConcurrencyErr = xerrors.NewSentinel("AsyncPush is called after Close")
 
-// PusherFromAsyncSink wraps the given sink into a (synchronous) pusher interface.
+// PusherFromAsyncSink wraps the given sink into a (synchronous) pusher interface
 func PusherFromAsyncSink(asink AsyncSink) Pusher {
 	return func(items []ChangeItem) error {
 		return <-asink.AsyncPush(items)

pkg/abstract/changeitem/change_item.go

@@ -211,10 +211,10 @@ func (c *ChangeItem) MakeMapKeys() map[string]bool {
 	return keyCols
 }
 
-// ColumnName is an explicitly declared column name.
+// ColumnName is an explicitly declared column name
 type ColumnName string
 
-// FastTableSchema is a mapping from column name to its schema.
+// FastTableSchema is a mapping from column name to its schema
 type FastTableSchema map[ColumnName]ColSchema
 
 // MakeFastTableSchema produces a fast table schema from an array of schemas of each column. Column names are taken from these schemas.
@@ -227,7 +227,7 @@ func MakeFastTableSchema(slowTableSchema []ColSchema) FastTableSchema {
 }
 
 // KeysChanged - checks if update changes primary keys
-// NOTE: this method is quite inefficient.
+// NOTE: this method is quite inefficient
 func (c *ChangeItem) KeysChanged() bool {
 	if c.Kind != UpdateKind {
 		return false
@@ -304,7 +304,7 @@ func (c *ChangeItem) IsToasted() bool {
 	return false
 }
 
-// OldOrCurrentKeys returns a string representing the values of the columns from the given set of columns, extracted from OldKeys or ColumnValues, if OldKeys are absent.
+// OldOrCurrentKeys returns a string representing the values of the columns from the given set of columns, extracted from OldKeys or ColumnValues, if OldKeys are absent
 func (c *ChangeItem) OldOrCurrentKeysString(keyColumns map[string]bool) string {
 	if (c.Kind == UpdateKind || c.Kind == DeleteKind) && len(c.OldKeys.KeyValues) > 0 {
 		keys := make(map[string]interface{})
@@ -360,7 +360,7 @@ func ContainsNonRowItem(s []ChangeItem) bool {
 	return false
 }
 
-// FindItemOfKind returns an item in the slice whose kind is among the given ones, or nil if there is no such item in it.
+// FindItemOfKind returns an item in the slice whose kind is among the given ones, or nil if there is no such item in it
 func FindItemOfKind(s []ChangeItem, kinds ...Kind) *ChangeItem {
 	wanted := make(map[Kind]bool)
 	for _, k := range kinds {
@@ -375,7 +375,7 @@ func FindItemOfKind(s []ChangeItem, kinds ...Kind) *ChangeItem {
 }
 
 // IsMirror
-// mirror - it's special format with hardcoded schema - used for "mirroring" (queue->queue) - transfer messages between queues without changes.
+// mirror - it's special format with hardcoded schema - used for "mirroring" (queue->queue) - transfer messages between queues without changes
 func (c *ChangeItem) IsMirror() bool {
 	if len(c.ColumnNames) != len(RawDataColumns) {
 		return false
@@ -683,7 +683,7 @@ func (c *ChangeItem) SetTableSchema(tableSchema *TableSchema) {
 }
 
 // RemoveColumns mutate change item to skip some columns from it.
-// it remove it from column names and column values.
+// it remove it from column names and column values
 func (c *ChangeItem) RemoveColumns(cols ...string) {
 	toDelete := map[string]struct{}{}
 	for _, col := range cols {

pkg/abstract/changeitem/kind.go

@@ -10,10 +10,10 @@ const (
 	// For each table being uploaded the following ChangeItems should be emited in order
 	// InitShardedTableLoad -> N * (InitTableLoad -> row events -> DoneTableLoad) -> DoneShardedTableLoad
 	// where N is number of table parts if table is sharded (thus N = 1 if table is not sharded)
-	// Each ChangeItem of those kinds should contain TableSchema for the corresponding table.
+	// Each ChangeItem of those kinds should contain TableSchema for the corresponding table
 
 	// InitShardedTableLoad is table header.
-	// All Init/DoneTableLoad and data items for the same table should be sent strictly after this kind of item.
+	// All Init/DoneTableLoad and data items for the same table should be sent strictly after this kind of item
 	InitShardedTableLoad = Kind("init_sharded_table_load")
 	// InitTableLoad is table part header. It should be sent for each table part before uploading any data for that part.
 	// If the table is not sharded, it is considered to have a single part so InitTableLoad should be sent

pkg/abstract/changeitem/table_id.go

@@ -48,7 +48,7 @@ func (t TableID) Equals(other TableID) bool {
 
 // Includes returns true if the given ID identifies a subset of tables identified by the current ID.
 //
-// In other words, it is a result of an operation "is a superset of".
+// In other words, it is a result of an operation "is a superset of"
 func (t TableID) Includes(sub TableID) bool {
 	if t.Equals(sub) {
 		return true

pkg/abstract/changeitem/table_part_id.go

@@ -14,7 +14,7 @@ type TablePartID struct {
 
 func (t *TablePartID) FqtnWithPartID() string {
 	if t.PartID == "" {
-		return t.Fqtn()
+		return t.TableID.Fqtn()
 	}
-	return fmt.Sprintf("%s.%s", t.Fqtn(), t.PartID)
+	return fmt.Sprintf("%s.%s", t.TableID.Fqtn(), t.PartID)
 }

pkg/abstract/changeitem/utils.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 )
 
-// MakeMapColNameToIndex returns a mapping of a column name to an index in the given slice.
+// MakeMapColNameToIndex returns a mapping of a column name to an index in the given slice
 func MakeMapColNameToIndex(in []ColSchema) map[string]int {
 	result := make(map[string]int)
 	for i, column := range in {
@@ -24,7 +24,7 @@ func KeyNames(in []ColSchema) []string {
 	return result
 }
 
-// Sniff dat data.
+// Sniff dat data
 func Sniff(input []ChangeItem) string {
 	tables := make(map[TableID]ChangeItem)
 	tablesCntr := make(map[TableID]int)

pkg/abstract/coordinator/editor.go

@@ -4,7 +4,7 @@ import "github.com/transferia/transferia/pkg/abstract/model"
 
 // Editor is obsolete interface to update entities on coordinator
 //
-// Deprecated: do not rely on it anymore.
+// Deprecated: do not rely on it anymore
 type Editor interface {
 	// GetEndpointTransfers get all *other* linked transfer to either source or target of provider *transferID*
 	GetEndpointTransfers(transferID string, isSource bool) ([]*model.Transfer, error)