Update with where clause

CHANGELOG.md

@@ -16,6 +16,8 @@ Types of changes
 
 ## [3.7.0]
 
+- `Added` flag `--where` (short `-w`) to `lino push update` command. This flag allows to specify a SQL WHERE clause to filter the rows to update.
+
 - `Added` flag `--commit-timeout` to `lino push` command. This flag allows to trigger a commit if no new row is received within the specified duration
 
 ## [3.6.2]

README.md

@@ -469,6 +469,27 @@ If a row with the same primary key already exists in the destination table, it w
 
 **Note:** This feature is currently supported for PostgreSQL and Oracle databases.
 
+### Update with where clause
+
+The `update` mode allows to update existing rows. You can use the `--where` flag (or `-w`) to specify a SQL WHERE clause. The update will only be performed if the condition matches.
+
+```bash
+$ lino push update dest --where "last_update > '2023-01-01'" < data.jsonl
+```
+
+The clause is added to the update query with an `AND` operator, combining it with the primary key condition.
+
+**Important:** The `--where` clause is applied **only** to the start table of the push operation. If the push operation involves multiple tables (via recursion in the ingress descriptor), the condition will not be applied to child or parent tables.
+
+#### Difference with `__usingpk__`
+
+* `__usingpk__` is used to target a specific row to update when the primary key itself is being modified. It provides the "old" primary key to locate the record.
+* `--where` is used to add a logical condition to the update. It is useful for:
+    * **Optimistic locking**: Update only if the record hasn't changed since last read (e.g. `version = 1`).
+    * **Business rules**: Update only if the record is in a specific state (e.g. `status = 'PENDING'`).
+
+Both can be used together. `__usingpk__` identifies *which* row to try to update, and `--where` adds an additional check to decide *if* the update should proceed.
+
 ### How to recover from error
 
 Use options `lino pull --exclude-from-file` (shortcut `-X`) and `lino push --savepoint` combined to handle error recovery. The process will restart where it failed if an error has interrupted it in a previous run.

internal/app/push/cli.go

@@ -94,7 +94,8 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
 		ingressDescriptor  string
 		rowExporter        push.RowWriter
 		pkTranslations     map[string]string
-		whereField         string
+		usingPkField       string
+		whereClause        string
 		savepoint          string
 		autoTruncate       bool
 		watch              bool
@@ -180,7 +181,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
 				observers = append(observers, observer)
 			}
 
-			e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, commitTimeout, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, observers...)
+			e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, commitTimeout, disableConstraints, rowExporter, translator, usingPkField, whereClause, savepoint, autoTruncate, observers...)
 			if e3 != nil {
 				log.Fatal().AnErr("error", e3).Msg("Fatal error stop the push command")
 				os.Exit(1)
@@ -200,11 +201,12 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
 	cmd.Flags().StringVarP(&table, "table", "t", "", "Table to writes json")
 	cmd.Flags().StringVarP(&ingressDescriptor, "ingress-descriptor", "i", "ingress-descriptor.yaml", "Ingress descriptor filename")
 	cmd.Flags().StringToStringVar(&pkTranslations, "pk-translation", map[string]string{}, "list of dictionaries old value / new value for primary key update")
-	cmd.Flags().StringVar(&whereField, "using-pk-field", "__usingpk__", "Name of the data field that can be used as pk for update queries")
+	cmd.Flags().StringVar(&usingPkField, "using-pk-field", "__usingpk__", "Name of the data field that can be used as pk for update queries")
 	cmd.Flags().StringVar(&savepoint, "savepoint", "", "Name of a file to write primary keys of effectively processed lines (commit to database)")
 	cmd.Flags().BoolVarP(&autoTruncate, "autotruncate", "a", false, "Automatically truncate values to the maximum length defined in table.yaml")
 	cmd.Flags().BoolVarP(&watch, "watch", "w", false, "watch statistics about pushed lines")
 	cmd.Flags().StringVarP(&logSQLTo, "log-sql", "l", "", "Log SQL requests and data to specified folder (1 file per table)")
+	cmd.Flags().StringVarP(&whereClause, "where", "W", "", "WHERE clause to add to the update query")
 	cmd.SetOut(out)
 	cmd.SetErr(err)
 	cmd.SetIn(in)

internal/app/push/http.go

@@ -146,7 +146,7 @@ func Handler(w http.ResponseWriter, r *http.Request, mode push.Mode, ingressDesc
 
 	log.Debug().Msg(fmt.Sprintf("call Push with mode %s", mode))
 
-	e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, 0, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", false)
+	e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, 0, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", "", false)
 	if e3 != nil {
 		log.Error().Err(e3).Msg("")
 		w.WriteHeader(http.StatusNotFound)

internal/infra/push/datadestination_http.go

@@ -68,7 +68,7 @@ func (dd *HTTPDataDestination) SafeUrl() string {
 }
 
 // Open HTTP Connection
-func (dd *HTTPDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool) *push.Error {
+func (dd *HTTPDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool, whereClause string) *push.Error {
 	log.Debug().Str("url", dd.url).Str("schema", dd.schema).Str("mode", mode.String()).Bool("disableConstraints", disableConstraints).Msg("open HTTP destination")
 	dd.mode = mode
 	dd.disableConstraints = disableConstraints

internal/infra/push/datadestination_sql.go

@@ -41,6 +41,8 @@ type SQLDataDestination struct {
 	disableConstraints bool
 	dialect            SQLDialect
 	sqlLogger          *SQLLogger
+	whereClause        string
+	startTableName     string
 }
 
 // NewSQLDataDestination creates a new SQL datadestination.
@@ -134,9 +136,13 @@ func (dd *SQLDataDestination) Commit() *push.Error {
 }
 
 // Open SQL Connection
-func (dd *SQLDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool) *push.Error {
+func (dd *SQLDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool, whereClause string) *push.Error {
 	dd.mode = mode
 	dd.disableConstraints = disableConstraints
+	dd.whereClause = whereClause
+	if plan.FirstTable() != nil {
+		dd.startTableName = plan.FirstTable().Name()
+	}
 
 	db, err := dburl.Open(dd.url)
 	if err != nil {
@@ -298,6 +304,9 @@ func (rw *SQLRowWriter) createStatement(row push.Row, where push.Row) *push.Erro
 		if pusherr != nil {
 			return pusherr
 		}
+		if rw.dd.whereClause != "" && rw.tableName() == rw.dd.startTableName {
+			prepareStmt += " AND (" + rw.dd.whereClause + ")"
+		}
 
 	case rw.dd.mode == push.Upsert:
 		prepareStmt, rw.headers, pusherr = rw.dd.dialect.UpsertStatement(rw.tableName(), selectValues, whereValues, rw.table.PrimaryKey())

internal/infra/push/datadestination_ws.go

@@ -134,7 +134,7 @@ func (dd *WebSocketDataDestination) SendMessageAndReadResult(msg CommandMessage)
 }
 
 // Open web socket connection
-func (dd *WebSocketDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool) *push.Error {
+func (dd *WebSocketDataDestination) Open(plan push.Plan, mode push.Mode, disableConstraints bool, whereClause string) *push.Error {
 	log.Debug().Str("url", dd.url).Str("schema", dd.schema).Str("mode", mode.String()).Bool("disableConstraints", disableConstraints).Msg("open web socket destination")
 	dd.mode = mode
 	dd.disableConstraints = disableConstraints

pkg/push/driven.go

@@ -24,7 +24,7 @@ type DataDestinationFactory interface {
 
 // DataDestination to write in the push process.
 type DataDestination interface {
-	Open(plan Plan, mode Mode, disableConstraints bool) *Error
+	Open(plan Plan, mode Mode, disableConstraints bool, whereClause string) *Error
 	Commit() *Error
 	RowWriter(table Table) (RowWriter, *Error)
 	OpenSQLLogger(folderPath string) error

pkg/push/driven_test.go

@@ -70,7 +70,7 @@ func (mdd *memoryDataDestination) OpenSQLLogger(string) error {
 	return nil
 }
 
-func (mdd *memoryDataDestination) Open(pla push.Plan, mode push.Mode, disableConstraints bool) *push.Error {
+func (mdd *memoryDataDestination) Open(pla push.Plan, mode push.Mode, disableConstraints bool, where string) *push.Error {
 	mdd.opened = true
 	return nil
 }

pkg/push/driver.go

@@ -33,6 +33,7 @@ type PushConfig struct {
 	CommitTimeout      time.Duration
 	DisableConstraints bool
 	WhereField         string
+	WhereClause        string
 	SavepointPath      string
 	AutoTruncate       bool
 }
@@ -52,12 +53,13 @@ type pushContext struct {
 }
 
 // Push write rows to target table
-func Push(ri RowIterator, destination DataDestination, plan Plan, mode Mode, commitSize uint, commitTimeout time.Duration, disableConstraints bool, catchError RowWriter, translator Translator, whereField string, savepointPath string, autotruncate bool, observers ...Observer) *Error {
+func Push(ri RowIterator, destination DataDestination, plan Plan, mode Mode, commitSize uint, commitTimeout time.Duration, disableConstraints bool, catchError RowWriter, translator Translator, whereField string, whereClause string, savepointPath string, autotruncate bool, observers ...Observer) *Error {
 	cfg := PushConfig{
 		CommitSize:         commitSize,
 		CommitTimeout:      commitTimeout,
 		DisableConstraints: disableConstraints,
 		WhereField:         whereField,
+		WhereClause:        whereClause,
 		SavepointPath:      savepointPath,
 		AutoTruncate:       autotruncate,
 	}
@@ -83,7 +85,7 @@ func (ctx *pushContext) Run(ri RowIterator) (err *Error) {
 		Str("url", ctx.destination.SafeUrl()).
 		Msg("Open database")
 
-	if err := ctx.destination.Open(ctx.plan, ctx.mode, ctx.cfg.DisableConstraints); err != nil {
+	if err := ctx.destination.Open(ctx.plan, ctx.mode, ctx.cfg.DisableConstraints, ctx.cfg.WhereClause); err != nil {
 		return err
 	}