【Bug】增量自动同步，etlCondition设置的 where 条件无效

[Bennie61]

• [x] I have checked the FAQ of this repository and believe that this is not a duplicate.

environment

• canal adapter version 1.1.7
• mysql version 8.0
• elasticsearch 7.14.0

Issue Description

mysql同步至es时，在canal adapter 适配器表映射文件中使用 etlCondition 添加where条件，增量自动同步时where条件不生效。 该问题在canal 1.1.5也存在，有同志曾提出同一问题但仍未解决。Issue 4092。 #4092

官方教程：

dataSourceKey: defaultDS        # 源数据源的key, 对应上面配置的srcDataSources中的值
outerAdapterKey: exampleKey     # 对应application.yml中es配置的key 
destination: example            # cannal的instance或者MQ的topic
groupId:                        # 对应MQ模式下的groupId, 只会同步对应groupId的数据
esMapping:
  _index: mytest_user           # es 的索引名称
  _type: _doc                   # es 的type名称, es7下无需配置此项
  _id: _id                      # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
#  pk: id                       # 如果不需要_id, 则需要指定一个属性为主键属性
  # sql映射
  sql: "select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name,
        a.c_time as _c_time, c.labels as _labels from user a
        left join role b on b.id=a.role_id
        left join (select user_id, group_concat(label order by id desc separator ';') as labels from label
        group by user_id) c on c.user_id=a.id"
#  objFields:
#    _labels: array:;           # 数组或者对象属性, array:; 代表以;字段里面是以;分隔的
#    _obj: object               # json对象
  etlCondition: "where a.c_time>='{0}'"     # etl 的条件参数
  commitBatch: 3000                         # 提交批大

那么在增量自动同步过程中 etlCondition 的where语句如何传参？？？ 感谢各位大佬赐教！

Steps to reproduce

适配器表映射文件 mytest_user.yml：

 sql: "SELECT 
    t.topic_id as _id,
    # 此处省略...
    t.type
FROM 
    topic t"
etlCondition: "where t.type=1"

Expected behaviour

只同步type值为1的数据至es

Actual behaviour

type不为1的数据也都同步至es中了

[jiandanfeng]

etlCondition应该只是用于etl的api接口

[Bennie61]

那增量自动同步没法进行过滤，只能同步全表？？我看官方教程(https://github.com/alibaba/canal/wiki/Sync-ES)里说“主sql中不能有where查询条件”。

[jiandanfeng]

可以使用 Elasticsearch 的 Ingest Pipelines，在数据索引之前对数据进行处理和过滤。

[ddl1228]

select a.id as _id, a.name as _name, a.role_id as _role_id, b.role_name as _role_name, a.c_time as _c_time, c.labels as _labels from user a left join role b on b.id=a.role_id left join (select user_id, group_concat(label order by id desc separator ';') as labels from label group by user_id) c on c.user_id=a.id 你这里关联了三张表，其中user是主表，当更新非主表，比如role表中的数据时，es上对应的字段值没有更新，只有更新主表user的数据时，es上对应的字段值才会更新，有遇到过这个问题吗？