Auto commit after timeout

CHANGELOG.md

@@ -14,6 +14,10 @@ Types of changes
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [3.7.0]
+
+- `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]
 
 - `Fixed` quoting column names in UPDATE during push (PostgreSQL)

README.md

@@ -345,6 +345,14 @@ Each line is a filter and `lino` apply it to the start table to extract data.
 
 The `push` sub-command import a **json** line stream (jsonline format http://jsonlines.org/) in each table, following the ingress descriptor defined in current directory.
 
+### Commit Timeout
+
+Use the `--commit-timeout` flag to specify a duration (e.g. `5s`, `1m`) after which a commit is triggered if no new row is received. This is useful for streaming applications where data might arrive slowly.
+
+```bash
+$ lino push insert dest --commit-timeout 5s
+```
+
 ### Autotruncate values
 
 Use the `autotruncate` flag to automatically truncate string values that overflows the maximum length accepted by the database.

internal/app/push/cli.go

@@ -99,6 +99,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
 		autoTruncate       bool
 		watch              bool
 		logSQLTo           string
+		commitTimeout      time.Duration
 	)
 
 	cmd := &cobra.Command{
@@ -179,7 +180,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, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, observers...)
+			e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, commitTimeout, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, observers...)
 			if e3 != nil {
 				log.Fatal().AnErr("error", e3).Msg("Fatal error stop the push command")
 				os.Exit(1)
@@ -193,6 +194,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra
 		},
 	}
 	cmd.Flags().UintVarP(&commitSize, "commitSize", "c", 500, "Commit size")
+	cmd.Flags().DurationVar(&commitTimeout, "commit-timeout", 0, "Commit timeout (e.g. 5s, 1m). If set, a commit is triggered if no new row is received within this duration.")
 	cmd.Flags().BoolVarP(&disableConstraints, "disable-constraints", "d", false, "Disable constraint during push")
 	cmd.Flags().StringVarP(&catchErrors, "catch-errors", "e", "", "Catch errors and write line in file")
 	cmd.Flags().StringVarP(&table, "table", "t", "", "Table to writes json")

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, 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_sql.go

@@ -382,8 +382,8 @@ func (rw *SQLRowWriter) Write(row push.Row, where push.Row) *push.Error {
 	rw.sqlLogger.Write(values)
 
 	_, err2 := rw.statement.Exec(values...)
-	log.Trace().AnErr("error", err2).Msg("push error")
 	if err2 != nil {
+		log.Trace().AnErr("error", err2).Msg("push error")
 		// reset statement after error
 		if err := rw.close(); err != nil {
 			return &push.Error{Description: err.Error() + "\noriginal error :\n" + err2.Error()}

pkg/push/driven_test.go

@@ -55,6 +55,7 @@ type memoryDataDestination struct {
 	closed    bool
 	committed bool
 	opened    bool
+	commits   int
 }
 
 func (mdd *memoryDataDestination) SafeUrl() string {
@@ -76,6 +77,7 @@ func (mdd *memoryDataDestination) Open(pla push.Plan, mode push.Mode, disableCon
 
 func (mdd *memoryDataDestination) Commit() *push.Error {
 	mdd.committed = true
+	mdd.commits++
 	return nil
 }