Repository: lf-edge/ekuiper
Pull request #2939 (2024-06-19 05:49:26)

feat(io): video source add props

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `91.66667%` with `6 lines` in your changes missing coverage. Please review.
> Project coverage is 69.70%. Comparing base [(`b664ac9`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/b664ac9903e5880a9ed169ae7297ae35f49297c7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`bbdb59a`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/bbdb59a2000236989e9859f6de55e6da4410ca51?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).
> Report is 2 commits behind head on master-v2.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/video/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?src=pr&el=tree&filepath=internal%2Fio%2Fvideo%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vdmlkZW8vc291cmNlLmdv) | 91.67% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [pkg/mock/test\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?src=pr&el=tree&filepath=pkg%2Fmock%2Ftest_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL21vY2svdGVzdF9zb3VyY2UuZ28=) | 91.67% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@              Coverage Diff              @@
##           master-v2    #2939      +/-   ##
=============================================
+ Coverage      69.51%   69.70%   +0.18%     
=============================================
  Files            301      302       +1     
  Lines          29112    29223     +111     
=============================================
+ Hits           20236    20367     +131     
+ Misses          7126     7104      -22     
- Partials        1750     1752       +2     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2939?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/go.mod b/go.mod
index b3e6db32..00203d21 100644
--- a/go.mod
+++ b/go.mod
@@ -38,6 +38,7 @@ require (
 	github.com/shirou/gopsutil v3.21.11+incompatible
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
+	github.com/u2takey/ffmpeg-go v0.5.0
 	github.com/ugorji/go/codec v1.2.12
 	github.com/urfave/cli v1.22.14
 	github.com/valyala/fastjson v1.6.4
@@ -58,6 +59,7 @@ require (
 	github.com/Masterminds/semver/v3 v3.2.1 // indirect
 	github.com/Microsoft/go-winio v0.6.1 // indirect
 	github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 // indirect
+	github.com/aws/aws-sdk-go v1.38.20 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bufbuild/protocompile v0.9.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -83,6 +85,7 @@ require (
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huandu/xstrings v1.4.0 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/jonboulle/clockwork v0.2.2 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/lestrrat-go/strftime v1.0.6 // indirect
@@ -111,6 +114,7 @@ require (
 	github.com/spf13/cast v1.6.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.13 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
+	github.com/u2takey/go-utils v0.3.1 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
diff --git a/go.sum b/go.sum
index 06413f79..7604e0f9 100644
--- a/go.sum
+++ b/go.sum
@@ -22,6 +22,8 @@ github.com/alicebob/miniredis/v2 v2.30.0 h1:uA3uhDbCxfO9+DI/DuGeAMr9qI+noVWwGPNT
 github.com/alicebob/miniredis/v2 v2.30.0/go.mod h1:84TWKZlxYkfgMucPBf5SOQBYJceZeQRFIaQgNMiCX6Q=
 github.com/apple/foundationdb/bindings/go v0.0.0-20240315202255-8943393e84fc h1:UDw1YBxVLq6vldL3ncJOtSgbnsG+2h4aHsDi9EFN/Bo=
 github.com/apple/foundationdb/bindings/go v0.0.0-20240315202255-8943393e84fc/go.mod h1:OMVSB21p9+xQUIqlGizHPZfjK+SHws1ht+ZytVDoz9U=
+github.com/aws/aws-sdk-go v1.38.20 h1:QbzNx/tdfATbdKfubBpkt84OM6oBkxQZRw6+bW2GyeA=
+github.com/aws/aws-sdk-go v1.38.20/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
 github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o=
 github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -51,6 +53,7 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/disintegration/imaging v1.6.2/go.mod h1:44/5580QXChDfwIclfc/PCwrr44amcmDAg8hxG0Ewe4=
 github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
 github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8=
 github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI=
@@ -84,6 +87,7 @@ github.com/gabriel-vasile/mimetype v1.4.3/go.mod h1:d8uq/6HKRL6CGdk+aubisF/M5GcP
 github.com/gdamore/optopia v0.2.0/go.mod h1:YKYEwo5C1Pa617H7NlPcmQXl+vG6YnSSNB44n8dNL0Q=
 github.com/gdexlab/go-render v1.0.1 h1:rxqB3vo5s4n1kF0ySmoNeSPRYkEsyHgln4jFIQY7v0U=
 github.com/gdexlab/go-render v1.0.1/go.mod h1:wRi5nW2qfjiGj4mPukH4UV0IknS1cHD4VgFTmJX5JzM=
+github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
@@ -101,6 +105,7 @@ github.com/go-sourcemap/sourcemap v2.1.3+incompatible/go.mod h1:F8jJfvm2KbVjc5Nq
 github.com/go-sourcemap/sourcemap v2.1.4+incompatible h1:a+iTbH5auLKxaNwQFg0B+TCYl6lbukKPc7b5x0n1s6Q=
 github.com/go-sourcemap/sourcemap v2.1.4+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
+github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
 github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3 h1:zN2lZNZRflqFyxVaTIU61KNKQ9C0055u9CAfpmqUvo4=
 github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3/go.mod h1:nPpo7qLxd6XL3hWJG/O60sR8ZKfMCiIoNap5GvD12KU=
 github.com/golang-jwt/jwt/v5 v5.2.1 h1:OuVbFODueb089Lh128TAcimifWaLhJwVflnrgM17wHk=
@@ -123,6 +128,7 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/pprof v0.0.0-20230207041349-798e818bf904/go.mod h1:uglQLonpP8qtYCYyzA+8c/9qtqgA3qsXGYqCPKARAFg=
 github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd h1:LjW4RcTwfcqOYGmD7UpFrn1gfBZ9mgu7QN5mSeFkCog=
 github.com/google/pprof v0.0.0-20240319011627-a57c5dfe54fd/go.mod h1:czg5+yv1E0ZGTi6S6vVK1mke0fV+FaUhNGcd6VRS9Ik=
@@ -141,6 +147,7 @@ github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
+github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
@@ -157,10 +164,17 @@ github.com/jinzhu/copier v0.3.5 h1:GlvfUwHk62RokgqVNvYsku0TATCF7bAHVwEXoBh3iJg=
 github.com/jinzhu/copier v0.3.5/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg=
 github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
 github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
+github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/jonboulle/clockwork v0.2.2 h1:UOGuzwb1PwsrDAObMuhUnj0p5ULPj8V/xJ7Kx9qUBdQ=
 github.com/jonboulle/clockwork v0.2.2/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1 h1:JL2rWnBX8jnbHHlLcLde3BBWs+jzqZvOmF+M3sXoNOE=
 github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1/go.mod h1:nNLjpEi4xVFB7358xLPpPscdvXP+pbhiHgSmjIur8z0=
+github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
 github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -196,6 +210,10 @@ github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zx
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/mochi-mqtt/server/v2 v2.6.4 h1:zuKokG/YzmefLecpodu1VSOSXJf1GP9mk2LdVcp1Jp4=
 github.com/mochi-mqtt/server/v2 v2.6.4/go.mod h1:TqztjKGO0/ArOjJt9x9idk0kqPT3CVN8Pb+l+PS5Gdo=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
@@ -227,12 +245,14 @@ github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7J
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.13.0 h1:7lLHu94wT9Ij0o6EWWclhu0aOh32VxhkwEJvzuWPeak=
 github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je41yGY=
+github.com/panjf2000/ants/v2 v2.4.2/go.mod h1:f6F0NZVFsGCp5A7QW/Zj/m92atWwOkY0OIhFxRNFr4A=
 github.com/pebbe/zmq4 v1.2.11 h1:Ua5mgIaZeabUGnH7tqswkUcjkL7JYGai5e8v4hpEU9Q=
 github.com/pebbe/zmq4 v1.2.11/go.mod h1:nqnPueOapVhE2wItZ0uOErngczsJdLOGkebMxaO8r48=
 github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
 github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c h1:CgbKAHto5CQgWM9fSBIvaxsJHuGP0uM74HXtv3MyyGQ=
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -269,6 +289,7 @@ github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFR
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cast v1.6.0 h1:GEiTHELF+vaR5dhz3VqZfFSzZjYbgeKDpBxQVS4GYJ0=
 github.com/spf13/cast v1.6.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
@@ -291,6 +312,10 @@ github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08
 github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
 github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr4=
 github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
+github.com/u2takey/ffmpeg-go v0.5.0 h1:r7d86XuL7uLWJ5mzSeQ03uvjfIhiJYvsRAJFCW4uklU=
+github.com/u2takey/ffmpeg-go v0.5.0/go.mod h1:ruZWkvC1FEiUNjmROowOAps3ZcWxEiOpFoHCvk97kGc=
+github.com/u2takey/go-utils v0.3.1 h1:TaQTgmEZZeDHQFYfd+AdUT1cT4QJgJn/XVPELhHw4ys=
+github.com/u2takey/go-utils v0.3.1/go.mod h1:6e+v5vEZ/6gu12w/DC2ixZdZtCrNokVxD0JUklcqdCs=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
 github.com/urfave/cli v1.22.14 h1:ebbhrRiGK2i4naQJr+1Xj92HXZCrK7MsyTS/ob3HnAk=
@@ -313,6 +338,7 @@ go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnw
 go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+gocv.io/x/gocv v0.25.0/go.mod h1:Rar2PS6DV+T4FL+PM535EImD/h13hGVaHhnCu1xarBs=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -322,6 +348,7 @@ golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
 golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
 golang.org/x/exp v0.0.0-20240318143956-a85f2c67cd81 h1:6R2FC06FonbXQ8pK11/PDFY6N6LWlf9KlzibaCapmqc=
 golang.org/x/exp v0.0.0-20240318143956-a85f2c67cd81/go.mod h1:CQ1k9gNrJ50XIzaKCRR2hssIjF07kZFEiieALBM/ARQ=
+golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
@@ -334,6 +361,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -357,6 +385,7 @@ golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200602225109-6fdc65e7d980/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -385,7 +414,9 @@ golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -427,6 +458,8 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
@@ -459,3 +492,4 @@ modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA=
 modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0=
 modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
 modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
diff --git a/internal/conf/yaml_config_storage.go b/internal/conf/yaml_config_storage.go
index 4bd0d5a5..5dcdc3c6 100644
--- a/internal/conf/yaml_config_storage.go
+++ b/internal/conf/yaml_config_storage.go
@@ -228,3 +228,19 @@ func GetCfgFromKVStorage(typ string, plugin string, confKey string) (map[string]
 	}
 	return kvStorage.GetByPrefix(key)
 }
+
+// ClearKVStorage only used in unit test
+func ClearKVStorage() error {
+	kvStorage, err := getKVStorage()
+	if err != nil {
+		return err
+	}
+	km, err := kvStorage.GetByPrefix("")
+	if err != nil {
+		return err
+	}
+	for key := range km {
+		kvStorage.Delete(key)
+	}
+	return nil
+}
diff --git a/internal/io/video/source.go b/internal/io/video/source.go
new file mode 100644
index 00000000..1c48c053
--- /dev/null
+++ b/internal/io/video/source.go
@@ -0,0 +1,116 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package video
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"net/url"
+	"os/exec"
+	"strings"
+	"time"
+
+	ffmpeg "github.com/u2takey/ffmpeg-go"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+const FRAMENUMBER = 5
+
+type Source struct {
+	Url string `json:"url"`
+	// Run ffmpeg -formats to get all supported format, default to 'image2'
+	Format string `json:"format"`
+	// Check https://www.ffmpeg.org/general.html#Video-Codecs, default to 'mjpeg'
+	Codec string `json:"codec"`
+	meta  map[string]any
+}
+
+func (s *Source) Provision(ctx api.StreamContext, props map[string]any) error {
+	c := exec.Command("ffmpeg", "-version")
+	output, err := c.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("check ffmpeg failed, err:%v", err)
+	}
+	ctx.GetLogger().Infof("ffmpeg dependency check result: %s", string(output))
+	if strings.Contains(string(output), "ffmpeg version") {
+		ctx.GetLogger().Infof("ffmpeg dependency check success")
+	} else {
+		return errors.New("ffmpeg dependency check failed")
+	}
+
+	s.Format = "image2"
+	s.Codec = "mjpeg"
+	err = cast.MapToStruct(props, s)
+	if err != nil {
+		return err
+	}
+
+	if s.Url == "" {
+		return errors.New("url is empty")
+	}
+	_, err = url.ParseRequestURI(s.Url)
+	if err != nil {
+		return fmt.Errorf("url is invalid, err:%v", err)
+	}
+	s.meta = map[string]any{"url": s.Url}
+	return nil
+}
+
+func (s *Source) Close(_ api.StreamContext) error {
+	// do nothing
+	return nil
+}
+
+func (s *Source) Connect(_ api.StreamContext) error {
+	return nil
+}
+
+func (s *Source) Pull(ctx api.StreamContext, trigger time.Time, ingest api.BytesIngest, ingestError api.ErrorIngest) {
+	buf, err := s.readFrameAsJpeg(ctx)
+	if err != nil {
+		ingestError(ctx, err)
+	} else {
+		ingest(ctx, buf.Bytes(), s.meta, trigger)
+	}
+}
+
+func (s *Source) readFrameAsJpeg(ctx api.StreamContext) (*bytes.Buffer, error) {
+	ctx.GetLogger().Debugf("read frame at %v", time.Now())
+	buf := bytes.NewBuffer(nil)
+	err := s.readTo(ctx, buf)
+	if err != nil {
+		return nil, fmt.Errorf("read frame failed, err:%v", err)
+	}
+	return buf, nil
+}
+
+func (s *Source) readTo(ctx api.StreamContext, out io.Writer) error {
+	ctx.GetLogger().Debugf("read frame at %v", time.Now())
+	return ffmpeg.Input(s.Url).
+		Filter("select", ffmpeg.Args{fmt.Sprintf("gte(n,%d)", FRAMENUMBER)}).
+		Output("pipe:", ffmpeg.KwArgs{"vframes": 1, "format": s.Format, "vcodec": s.Codec}).
+		WithOutput(out).
+		Run()
+}
+
+func GetSource() api.Source {
+	return &Source{}
+}
+
+var _ api.PullBytesSource = &Source{}
diff --git a/internal/server/bump/bump3.go b/internal/server/bump/bump3.go
new file mode 100644
index 00000000..f6309fb8
--- /dev/null
+++ b/internal/server/bump/bump3.go
@@ -0,0 +1,146 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/store"
+)
+
+func bumpFrom2TO3() error {
+	return rewriteSQLSourceConfiguration()
+}
+
+func rewriteSQLSourceConfiguration() error {
+	keyProps, err := conf.GetCfgFromKVStorage("sources", "sql", "")
+	if err != nil {
+		return err
+	}
+	for key, props := range keyProps {
+		cfg := &OriginSqlSourceCfg{}
+		if err := cast.MapToStruct(props, cfg); err != nil || cfg == nil {
+			continue
+		}
+		data := rewriteCfg(cfg)
+		_, _, confKey, _ := extractKey(key)
+		conf.WriteCfgIntoKVStorage("sources", "sql", confKey, data)
+	}
+	return nil
+}
+
+func rewriteCfg(cfg *OriginSqlSourceCfg) map[string]interface{} {
+	m := make(map[string]interface{})
+	m["dburl"] = cfg.Url
+	m["interval"] = time.Duration(cfg.Interval).String()
+	if cfg.InternalSqlQueryCfg != nil {
+		icfg := cfg.InternalSqlQueryCfg
+		im := make(map[string]interface{})
+		im["table"] = icfg.Table
+		im["limit"] = icfg.Limit
+		f := extractIndexField(icfg)
+		if f != nil {
+			icfg.IndexFields = append(icfg.IndexFields, f)
+		}
+		if len(icfg.IndexFields) > 0 {
+			im["indexFields"] = icfg.IndexFields
+		}
+		m["internalSqlQueryCfg"] = im
+	}
+	if cfg.TemplateSqlQueryCfg != nil {
+		tcfg := cfg.TemplateSqlQueryCfg
+		tm := make(map[string]interface{})
+		tm["templateSql"] = tcfg.TemplateSQL
+		f := extractIndexField2(tcfg)
+		if f != nil {
+			tcfg.IndexFields = append(tcfg.IndexFields, f)
+		}
+		if len(tcfg.IndexFields) > 0 {
+			tm["indexFields"] = tcfg.IndexFields
+		}
+		m["templateSqlQueryCfg"] = tm
+	}
+	bs, _ := json.Marshal(m)
+	newm := make(map[string]interface{})
+	json.Unmarshal(bs, &newm)
+	return newm
+}
+
+func extractIndexField(icfg *OriginInternalSqlQueryCfg) *store.IndexField {
+	if len(icfg.IndexFieldName) < 1 {
+		return nil
+	}
+	f := &store.IndexField{
+		IndexFieldName: icfg.IndexFieldName,
+	}
+	if icfg.IndexFieldValue != nil {
+		f.IndexFieldValue = icfg.IndexFieldValue
+	}
+	if len(icfg.IndexFieldDataType) > 0 {
+		f.IndexFieldDataType = icfg.IndexFieldDataType
+	}
+	if len(icfg.IndexFieldDateTimeFormat) > 0 {
+		f.IndexFieldDateTimeFormat = icfg.IndexFieldDateTimeFormat
+	}
+	return f
+}
+
+func extractIndexField2(icfg *OriginTemplateSqlQueryCfg) *store.IndexField {
+	if len(icfg.IndexFieldName) < 1 {
+		return nil
+	}
+	f := &store.IndexField{
+		IndexFieldName: icfg.IndexFieldName,
+	}
+	if icfg.IndexFieldValue != nil {
+		f.IndexFieldValue = icfg.IndexFieldValue
+	}
+	if len(icfg.IndexFieldDataType) > 0 {
+		f.IndexFieldDataType = icfg.IndexFieldDataType
+	}
+	if len(icfg.IndexFieldDateTimeFormat) > 0 {
+		f.IndexFieldDateTimeFormat = icfg.IndexFieldDateTimeFormat
+	}
+	return f
+}
+
+type OriginSqlSourceCfg struct {
+	Url                 string                     `json:"url"`
+	Interval            cast.DurationConf          `json:"interval"`
+	InternalSqlQueryCfg *OriginInternalSqlQueryCfg `json:"internalSqlQueryCfg,omitempty"`
+	TemplateSqlQueryCfg *OriginTemplateSqlQueryCfg `json:"templateSqlQueryCfg,omitempty"`
+}
+
+type OriginInternalSqlQueryCfg struct {
+	Table                    string              `json:"table"`
+	Limit                    int                 `json:"limit"`
+	IndexFieldName           string              `json:"indexField"`
+	IndexFieldValue          interface{}         `json:"indexValue"`
+	IndexFieldDataType       string              `json:"indexFieldType"`
+	IndexFieldDateTimeFormat string              `json:"dateTimeFormat"`
+	IndexFields              []*store.IndexField `json:"indexFields"`
+}
+
+type OriginTemplateSqlQueryCfg struct {
+	TemplateSQL              string              `json:"templateSql"`
+	IndexFieldName           string              `json:"indexField"`
+	IndexFieldValue          interface{}         `json:"indexValue"`
+	IndexFieldDataType       string              `json:"indexFieldType"`
+	IndexFieldDateTimeFormat string              `json:"dateTimeFormat"`
+	IndexFields              []*store.IndexField `json:"indexFields"`
+}
diff --git a/internal/server/bump/bump_manager.go b/internal/server/bump/bump_manager.go
index a760aad5..b85ed3e0 100644
--- a/internal/server/bump/bump_manager.go
+++ b/internal/server/bump/bump_manager.go
@@ -30,7 +30,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/kv"
 )
 
-const currentVersion = 2
+const currentVersion = 3
 
 var GlobalBumpManager *BumpManager
 
@@ -81,6 +81,10 @@ func BumpToCurrentVersion(dataDir string) error {
 			if err := bumpFrom1TO2(); err != nil {
 				return err
 			}
+		case 3:
+			if err := bumpFrom2TO3(); err != nil {
+				return err
+			}
 		}
 	}
 	return nil


Test patch:
diff --git a/.github/workflows/run_test_case.yaml b/.github/workflows/run_test_case.yaml
index 0dc38ffa..18691caa 100644
--- a/.github/workflows/run_test_case.yaml
+++ b/.github/workflows/run_test_case.yaml
@@ -32,7 +32,7 @@ jobs:
       env:
         DEBIAN_FRONTEND: noninteractive
       run: |
-        pip3 install pynng==0.7.2 && sudo apt-get update && sudo apt-get install libzmq3-dev -y
+        pip3 install pynng==0.7.2 && sudo apt-get update && sudo apt-get install ffmpeg libzmq3-dev -y
     - name: Build plugins
       run: |
         curl -sSf https://raw.githubusercontent.com/WasmEdge/WasmEdge/master/utils/install.sh | bash -s -- -v 0.13.4 > /dev/null
diff --git a/internal/io/video/source_test.go b/internal/io/video/source_test.go
new file mode 100644
index 00000000..c8ace930
--- /dev/null
+++ b/internal/io/video/source_test.go
@@ -0,0 +1,103 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package video
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestProvision(t *testing.T) {
+	tests := []struct {
+		name  string
+		props map[string]any
+		err   string
+	}{
+		{
+			name: "wrong param type",
+			props: map[string]any{
+				"url": 45,
+			},
+			err: "",
+		},
+		{
+			name:  "missing url",
+			props: map[string]any{},
+			err:   "missing url",
+		},
+		{
+			name: "wrong url",
+			props: map[string]any{
+				"url": "rtsp/dafsa",
+			},
+			err: "wrong url",
+		},
+	}
+	ctx := mockContext.NewMockContext("test", "test")
+	s := GetSource()
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := s.Provision(ctx, test.props)
+			assert.Error(t, err)
+		})
+	}
+}
+
+func TestPull(t *testing.T) {
+	meta := map[string]any{
+		"url": "https://gcwbcdks.v.kcdnvip.com/gcwbcd/cdrmipanda_1/index.m3u8",
+	}
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple(nil, meta, timex.GetNow()),
+	}
+	r := GetSource()
+	mock.TestSourceConnectorCompare(t, r, map[string]any{
+		"url":      "https://gcwbcdks.v.kcdnvip.com/gcwbcd/cdrmipanda_1/index.m3u8",
+		"interval": "15s",
+	}, exp, func(e any, r any) bool {
+		et, ok := e.([]api.MessageTuple)
+		b := assert.True(t, ok)
+		rt, ok := r.([]api.MessageTuple)
+		b = b && assert.True(t, ok, "result is not []api.MessageTuple")
+		b = b && assert.Equal(t, len(et), len(rt))
+		for i := range et {
+			rti := rt[i].(*model.DefaultSourceTuple)
+			b = b && assert.True(t, len(rti.Raw()) > 100)
+		}
+		return b
+	}, func() {
+		// do nothing
+	})
+}
+
+func TestPullError(t *testing.T) {
+	exp := errors.New("read frame failed, err:exit status 1")
+	r := GetSource()
+	mock.TestSourceConnector(t, r, map[string]any{
+		"url":      "https://gcwbcdks.v.kcdnvip.com/gcwbcd/cdrmipanda_1/index.m3u8",
+		"codec":    "ogv",
+		"interval": "15s",
+	}, exp, func() {
+		// do nothing
+	})
+}
diff --git a/internal/server/bump/bump3_test.go b/internal/server/bump/bump3_test.go
new file mode 100644
index 00000000..beddaa64
--- /dev/null
+++ b/internal/server/bump/bump3_test.go
@@ -0,0 +1,202 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bump
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/store"
+)
+
+func TestRewriteSQLConf(t *testing.T) {
+	testcases := []struct {
+		oldCfg   *OriginSqlSourceCfg
+		expected *TargetSqlSourceCfg
+	}{
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &OriginInternalSqlQueryCfg{
+					Table:              "t",
+					Limit:              10,
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &TargetInternalSqlQueryCfg{
+					Table: "t",
+					Limit: 10,
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				TemplateSqlQueryCfg: &OriginTemplateSqlQueryCfg{
+					TemplateSQL:        "select * from t",
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				TemplateSqlQueryCfg: &TargetTemplateSqlQueryCfg{
+					TemplateSQL: "select * from t",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+		{
+			oldCfg: &OriginSqlSourceCfg{
+				Url:      "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &OriginInternalSqlQueryCfg{
+					Table:              "t",
+					Limit:              10,
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    1,
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+				TemplateSqlQueryCfg: &OriginTemplateSqlQueryCfg{
+					TemplateSQL:        "select * from t",
+					IndexFieldName:     "a",
+					IndexFieldValue:    1,
+					IndexFieldDataType: "bigint",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    1,
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+			expected: &TargetSqlSourceCfg{
+				DbUrl:    "123",
+				Interval: cast.DurationConf(10 * time.Second),
+				InternalSqlQueryCfg: &TargetInternalSqlQueryCfg{
+					Table: "t",
+					Limit: 10,
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+				TemplateSqlQueryCfg: &TargetTemplateSqlQueryCfg{
+					TemplateSQL: "select * from t",
+					IndexFields: []*store.IndexField{
+						{
+							IndexFieldName:     "b",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+						{
+							IndexFieldName:     "a",
+							IndexFieldValue:    float64(1),
+							IndexFieldDataType: "bigint",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for index, tc := range testcases {
+		require.NoError(t, conf.ClearKVStorage())
+		originData, err := json.Marshal(tc.oldCfg)
+		require.NoError(t, err)
+		m := make(map[string]interface{})
+		require.NoError(t, json.Unmarshal(originData, &m))
+		cfgKey := fmt.Sprintf("%v", index)
+		require.NoError(t, conf.WriteCfgIntoKVStorage("sources", "sql", cfgKey, m))
+		require.NoError(t, rewriteSQLSourceConfiguration())
+		gotm, err := conf.GetCfgFromKVStorage("sources", "sql", cfgKey)
+		require.NoError(t, err)
+		tcfg := &TargetSqlSourceCfg{}
+		cast.MapToStruct(gotm[fmt.Sprintf("sources.sql.%s", cfgKey)], tcfg)
+		require.Equal(t, tc.expected, tcfg)
+	}
+}
+
+type TargetSqlSourceCfg struct {
+	DbUrl               string                     `json:"dburl"`
+	Interval            cast.DurationConf          `json:"interval"`
+	InternalSqlQueryCfg *TargetInternalSqlQueryCfg `json:"internalSqlQueryCfg,omitempty"`
+	TemplateSqlQueryCfg *TargetTemplateSqlQueryCfg `json:"templateSqlQueryCfg,omitempty"`
+}
+
+type TargetInternalSqlQueryCfg struct {
+	Table       string              `json:"table"`
+	Limit       int                 `json:"limit"`
+	IndexFields []*store.IndexField `json:"indexFields"`
+}
+
+type TargetTemplateSqlQueryCfg struct {
+	TemplateSQL string              `json:"templateSql"`
+	IndexFields []*store.IndexField `json:"indexFields"`
+}
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index fb366d86..ff16615f 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -39,7 +39,7 @@ type ic struct {
 	IgnoreTs bool          `json:"ignoreTs"`
 }
 
-func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expected any, sender func()) {
+func TestSourceConnectorCompare(t *testing.T, r api.Source, props map[string]any, expected any, compare func(expected, result any) bool, sender func()) {
 	// init
 	c := count.Load()
 	if c == nil {
@@ -69,14 +69,25 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		limit = len(et)
 	case []api.RawTuple:
 		limit = len(et)
+	case error:
+		limit = 1
 	default:
 		t.Fatal("invalid expected type")
 	}
 	var (
 		wg     sync.WaitGroup
 		result []api.MessageTuple
+		e      error
 	)
 	wg.Add(1)
+	ingestErr := func(ctx api.StreamContext, err error) {
+		log.Println(err)
+		e = err
+		limit--
+		if limit == 0 {
+			wg.Done()
+		}
+	}
 	ingestBytes := func(ctx api.StreamContext, payload []byte, meta map[string]any, ts time.Time) {
 		if cc.IgnoreTs {
 			result = append(result, model.NewDefaultRawTupleIgnoreTs(payload, meta))
@@ -112,23 +123,15 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 	go func() {
 		switch ss := r.(type) {
 		case api.BytesSource:
-			err = ss.Subscribe(ctx, ingestBytes, func(ctx api.StreamContext, err error) {
-				log.Println(err)
-			})
+			err = ss.Subscribe(ctx, ingestBytes, ingestErr)
 		case api.TupleSource:
-			err = ss.Subscribe(ctx, ingestTuples, func(ctx api.StreamContext, err error) {
-				panic(err)
-			})
+			err = ss.Subscribe(ctx, ingestTuples, ingestErr)
 		case api.PullBytesSource, api.PullTupleSource:
 			switch ss := r.(type) {
 			case api.PullBytesSource:
-				ss.Pull(ctx, timex.GetNow(), ingestBytes, func(ctx api.StreamContext, err error) {
-					panic(err)
-				})
+				ss.Pull(ctx, timex.GetNow(), ingestBytes, ingestErr)
 			case api.PullTupleSource:
-				ss.Pull(ctx, timex.GetNow(), ingestTuples, func(ctx api.StreamContext, err error) {
-					panic(err)
-				})
+				ss.Pull(ctx, timex.GetNow(), ingestTuples, ingestErr)
 			}
 			ticker := timex.GetTicker(cc.Interval)
 			go func() {
@@ -139,13 +142,9 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 						ctx.GetLogger().Debugf("source pull at %v", tc.UnixMilli())
 						switch ss := r.(type) {
 						case api.PullBytesSource:
-							ss.Pull(ctx, tc, ingestBytes, func(ctx api.StreamContext, err error) {
-								panic(err)
-							})
+							ss.Pull(ctx, tc, ingestBytes, ingestErr)
 						case api.PullTupleSource:
-							ss.Pull(ctx, tc, ingestTuples, func(ctx api.StreamContext, err error) {
-								panic(err)
-							})
+							ss.Pull(ctx, tc, ingestTuples, ingestErr)
 						}
 					case <-ctx.Done():
 						return
@@ -171,11 +170,20 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 	case <-ctx.Done():
 	case <-finished:
 		cancel()
-		assert.Equal(t, expected, result)
 	case <-ticker:
 		cancel()
 		assert.Fail(t, "timeout")
 		return
 	}
-	assert.Equal(t, expected, result)
+	if ee, ok := expected.(error); ok {
+		assert.Equal(t, ee, e)
+	} else {
+		assert.True(t, compare(expected, result))
+	}
+}
+
+func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expected any, sender func()) {
+	TestSourceConnectorCompare(t, r, props, expected, func(expected, result any) bool {
+		return assert.Equal(t, expected, result)
+	}, sender)
 }