fix(sink): report io error when write to tdengine/sql error (#1345)

Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>

extensions/sinks/sql/sql.go

@@ -21,6 +21,7 @@ import (
 	"github.com/lf-edge/ekuiper/extensions/sqldatabase/driver"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 	"github.com/xo/dburl"
 	"reflect"
 	"strings"
@@ -109,7 +110,7 @@ func (m *sqlSink) writeToDB(ctx api.StreamContext, sqlStr *string) error {
 	ctx.GetLogger().Debugf(*sqlStr)
 	rows, err := m.db.Query(*sqlStr)
 	if err != nil {
-		return err
+		return fmt.Errorf("%s: %s", errorx.IOErr, err.Error())
 	}
 	return rows.Close()
 }

extensions/sinks/tdengine/tdengine.go

@@ -21,6 +21,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/errorx"
 	_ "github.com/taosdata/driver-go/v2/taosSql"
 	"reflect"
 	"strings"
@@ -268,7 +269,7 @@ func (m *taosSink) writeToDB(ctx api.StreamContext, SqlVal *string) error {
 	ctx.GetLogger().Debugf(finalSql)
 	rows, err := m.db.Query(finalSql)
 	if err != nil {
-		return err
+		return fmt.Errorf("%s: %s", errorx.IOErr, err.Error())
 	}
 	rows.Close()
 	return nil