refactor: replace hardcoded column names with reserved fields

Author: flarco

core/dbio/iop/duckdb.go

@@ -1426,8 +1426,7 @@ func (duck *DuckDb) MakeScanQuery(format dbio.FileType, uri string, fsc FileStre
 	}
 
 	// reserved word to use for timestamp comparison (when listing)
-	const slingLoadedAtColumn = "_sling_loaded_at"
-	if fsc.IncrementalKey != "" && fsc.IncrementalKey != slingLoadedAtColumn &&
+	if fsc.IncrementalKey != "" && fsc.IncrementalKey != env.ReservedFields.LoadedAt &&
 		fsc.IncrementalValue != "" {
 		incrementalWhereCond = g.F("%s > %s", dbio.TypeDbDuckDb.Quote(fsc.IncrementalKey), fsc.IncrementalValue)
 		where = g.F("where %s", incrementalWhereCond)

core/env/env.go

@@ -49,9 +49,23 @@ var (
 	setupOtel = func() {}
 
 	ReservedFields = struct {
-		SyncedOp string
+		SyncedOp  string
+		LoadedAt  string
+		SyncedAt  string
+		DeletedAt string
+		StreamURL string
+		RowNum    string
+		RowID     string
+		ExecID    string
 	}{
-		SyncedOp: "_sling_synced_op",
+		SyncedOp:  "_sling_synced_op",
+		LoadedAt:  "_sling_loaded_at",
+		SyncedAt:  "_sling_synced_at",
+		DeletedAt: "_sling_deleted_at",
+		StreamURL: "_sling_stream_url",
+		RowNum:    "_sling_row_num",
+		RowID:     "_sling_row_id",
+		ExecID:    "_sling_exec_id",
 	}
 )
 

core/sling/config.go

@@ -15,6 +15,7 @@ import (
 	"github.com/slingdata-io/sling-cli/core/dbio/connection"
 	"github.com/slingdata-io/sling-cli/core/dbio/database"
 	"github.com/slingdata-io/sling-cli/core/dbio/filesys"
+	"github.com/slingdata-io/sling-cli/core/env"
 	"github.com/spf13/cast"
 
 	"github.com/flarco/g"
@@ -161,7 +162,7 @@ func (cfg *Config) SetDefault() {
 	if val := os.Getenv("SLING_SYNCED_AT_COLUMN"); val != "" {
 		if cast.ToBool(val) {
 			cfg.MetadataSyncedAt = g.Bool(true)
-			slingDeletedAtColumn = slingSyncedAtColumn // deleted_at becomes synched_at
+			env.ReservedFields.DeletedAt = env.ReservedFields.SyncedAt // deleted_at becomes synched_at
 		} else {
 			cfg.MetadataSyncedAt = g.Bool(false)
 		}
@@ -366,10 +367,10 @@ func (cfg *Config) DetermineType() (Type JobType, err error) {
 			// OK, no need for update key
 		} else if srcApiProvided {
 			// OK, no need for update key/pk, API uses SLING_STATE for tracking
-		} else if srcFileProvided && cfg.Source.UpdateKey == slingLoadedAtColumn {
+		} else if srcFileProvided && cfg.Source.UpdateKey == env.ReservedFields.LoadedAt {
 			// need to loaded_at column for file incremental
 			cfg.MetadataLoadedAt = g.Bool(true)
-		} else if srcFileProvided && cfg.Source.UpdateKey == slingSyncedAtColumn {
+		} else if srcFileProvided && cfg.Source.UpdateKey == env.ReservedFields.SyncedAt {
 			cfg.MetadataSyncedAt = g.Bool(true)
 		} else if cfg.Source.UpdateKey == "" && len(cfg.Source.PrimaryKey()) == 0 {
 			err = g.Error("must specify value for 'update_key' and/or 'primary_key' for incremental mode. See docs for more details: https://docs.slingdata.io/sling-cli/run/configuration")

core/sling/task.go

@@ -342,33 +342,33 @@ func (t *TaskExecution) GetRate(secWindow int) (rowRate, byteRate int64) {
 
 func (t *TaskExecution) setGetMetadata() (metadata iop.Metadata) {
 	if t.Config.MetadataSyncedAt != nil && *t.Config.MetadataSyncedAt {
-		metadata.SyncedAt.Key = slingSyncedAtColumn
+		metadata.SyncedAt.Key = env.ReservedFields.SyncedAt
 		metadata.SyncedAt.Value = *t.StartTime // only timestamp
 		metadata.SyncedOp.Key = env.ReservedFields.SyncedOp
 		metadata.SyncedOp.Value = "I" // default to insert operation
 	} else if t.Config.MetadataLoadedAt != nil && *t.Config.MetadataLoadedAt {
-		metadata.SyncedAt.Key = slingLoadedAtColumn
+		metadata.SyncedAt.Key = env.ReservedFields.LoadedAt
 		if os.Getenv("SLING_LOADED_AT_COLUMN") == "timestamp" {
 			metadata.SyncedAt.Value = *t.StartTime
 		} else {
 			metadata.SyncedAt.Value = t.StartTime.Unix()
 		}
 	}
 	if t.Config.MetadataStreamURL {
-		metadata.StreamURL.Key = slingStreamURLColumn
+		metadata.StreamURL.Key = env.ReservedFields.StreamURL
 	}
 
 	if t.Config.MetadataRowID {
-		metadata.RowID.Key = slingRowIDColumn
+		metadata.RowID.Key = env.ReservedFields.RowID
 	}
 
 	if t.Config.MetadataExecID {
-		metadata.ExecID.Key = slingExecIDColumn
+		metadata.ExecID.Key = env.ReservedFields.ExecID
 		metadata.ExecID.Value = t.ExecID
 	}
 
 	if t.Config.MetadataRowNum {
-		metadata.RowNum.Key = slingRowNumColumn
+		metadata.RowNum.Key = env.ReservedFields.RowNum
 	}
 
 	// StarRocks: add _sling_row_id column if there is no primary,
@@ -390,8 +390,8 @@ func (t *TaskExecution) setGetMetadata() (metadata iop.Metadata) {
 		}
 
 		if addRowIDCol {
-			metadata.RowID.Key = slingRowIDColumn
-			t.Config.Target.Options.TableKeys[iop.HashKey] = []string{slingRowIDColumn}
+			metadata.RowID.Key = env.ReservedFields.RowID
+			t.Config.Target.Options.TableKeys[iop.HashKey] = []string{env.ReservedFields.RowID}
 		}
 	}
 

core/sling/task_run.go

@@ -25,14 +25,7 @@ import (
 )
 
 var (
-	start                time.Time
-	slingLoadedAtColumn  = "_sling_loaded_at"
-	slingSyncedAtColumn  = "_sling_synced_at"
-	slingDeletedAtColumn = "_sling_deleted_at"
-	slingStreamURLColumn = "_sling_stream_url"
-	slingRowNumColumn    = "_sling_row_num"
-	slingRowIDColumn     = "_sling_row_id"
-	slingExecIDColumn    = "_sling_exec_id"
+	start time.Time
 )
 
 var deleteMissing func(*TaskExecution, database.Connection, database.Connection) error = func(_ *TaskExecution, _, _ database.Connection) error {
@@ -484,7 +477,7 @@ func (t *TaskExecution) runFileToDB() (err error) {
 		t.Context.Map.Set("incremental_value", t.Config.IncrementalValStr)
 	} else if t.isIncrementalWithUpdateKey() && !t.Config.IsIncrementalWithRange() {
 		if t.Config.Source.UpdateKey == "." {
-			t.Config.Source.UpdateKey = slingLoadedAtColumn
+			t.Config.Source.UpdateKey = env.ReservedFields.LoadedAt
 		}
 		t.SetProgress("getting checkpoint value (%s)", t.Config.Source.UpdateKey)
 

core/sling/task_run_read.go

@@ -12,6 +12,7 @@ import (
 	"github.com/slingdata-io/sling-cli/core/dbio/database"
 	"github.com/slingdata-io/sling-cli/core/dbio/filesys"
 	"github.com/slingdata-io/sling-cli/core/dbio/iop"
+	"github.com/slingdata-io/sling-cli/core/env"
 	"github.com/spf13/cast"
 )
 
@@ -273,7 +274,7 @@ func (t *TaskExecution) ReadFromFile(cfg *Config) (df *iop.Dataflow, err error)
 
 	if t.Config.HasIncrementalVal() && !t.Config.IsFileStreamWithStateAndParts() {
 		// file stream incremental mode
-		if g.In(t.Config.Source.UpdateKey, slingLoadedAtColumn, slingSyncedAtColumn) {
+		if g.In(t.Config.Source.UpdateKey, env.ReservedFields.LoadedAt, env.ReservedFields.SyncedAt) {
 			options["SLING_FS_TIMESTAMP"] = t.Config.IncrementalValStr
 			g.Debug(`file stream using file_sys_timestamp=%#v and update_key=%s`, t.Config.IncrementalValStr, t.Config.Source.UpdateKey)
 		} else {