# KCQL function doesn't work in influx db sink connector

**URL:** <https://ask.lenses.io/t/kcql-function-doesnt-work-in-influx-db-sink-connector/43>\
**Category:** Apache Kafka Connectors\
**Created:** [April 17, 2023, 3:26am UTC](https://ask.lenses.io/t/kcql-function-doesnt-work-in-influx-db-sink-connector/43 "2023-04-17T03:26:26Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![omalcooray](https://dub1.discourse-cdn.com/flex017/user_avatar/ask.lenses.io/omalcooray/32/65_2.png) [@omalcooray](https://ask.lenses.io/u/omalcooray)\
**Post date:** [April 17, 2023, 3:26am UTC](https://ask.lenses.io/t/kcql-function-doesnt-work-in-influx-db-sink-connector/43/1 "2023-04-17T03:26:26Z")

</div>

When I’m trying to use following query on influx db sink connector I’m getting below error  
Query:  
INSERT INTO orders\_model1 SELECT id, name, order\_code, COALESCE(order\_type, ‘’) AS order\_type FROM orders WITHTIMESTAMP sys\_time()

Error

```auto
java.lang.IllegalArgumentException: Invalid syntax.failed to parse at line 1 due to mismatched input '(' expecting FROM
              at com.datamountaineer.kcql.Kcql.parse(Kcql.java:897)
              at com.datamountaineer.streamreactor.connect.config.Helpers$$anonfun$3.apply(Helpers.scala:101)
              at com.datamountaineer.streamreactor.connect.config.Helpers$$anonfun$3.apply(Helpers.scala:101)
              at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
              at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
              at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
              at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
              at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
              at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
              at com.datamountaineer.streamreactor.connect.config.Helpers$.checkInputTopics(Helpers.scala:101)
              at com.datamountaineer.streamreactor.connect.influx.InfluxSinkConnector.start(InfluxSinkConnector.scala:65)
              at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:184)
              at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:209)
              at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:348)
              at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:331)
              at org.apache.kafka.connect.runtime.WorkerConnector.doRun(WorkerConnector.java:140)
              at org.apache.kafka.connect.runtime.WorkerConnector.run(WorkerConnector.java:117)
              at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
              at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
              at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
              at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
              at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.IllegalStateException: failed to parse at line 1 due to mismatched input '(' expecting FROM
  at com.datamountaineer.kcql.Kcql$1.syntaxError(Kcql.java:403)
  at org.antlr.v4.runtime.ProxyErrorListener.syntaxError(ProxyErrorListener.java:41)
  at org.antlr.v4.runtime.Parser.notifyErrorListeners(Parser.java:544)
  at org.antlr.v4.runtime.DefaultErrorStrategy.reportInputMismatch(DefaultErrorStrategy.java:299)
  at org.antlr.v4.runtime.DefaultErrorStrategy.reportError(DefaultErrorStrategy.java:124)
  at com.datamountaineer.kcql.antlr4.ConnectorParser.select_clause_basic(ConnectorParser.java:1436)
  at com.datamountaineer.kcql.antlr4.ConnectorParser.insert_from_clause(ConnectorParser.java:704)
  at com.datamountaineer.kcql.antlr4.ConnectorParser.stat(ConnectorParser.java:207)
  at com.datamountaineer.kcql.Kcql.parse(Kcql.java:895)
  ... 21 more
Caused by: org.antlr.v4.runtime.InputMismatchException
  at org.antlr.v4.runtime.DefaultErrorStrategy.recoverInline(DefaultErrorStrategy.java:453)
  at org.antlr.v4.runtime.Parser.match(Parser.java:206)
  at com.datamountaineer.kcql.antlr4.ConnectorParser.select_clause_basic(ConnectorParser.java:1419)
  ... 24 more

```

---

<div class="post-metadata">

**Author:** ![stheppi](https://dub1.discourse-cdn.com/flex017/user_avatar/ask.lenses.io/stheppi/32/30_2.png) [@stheppi](https://ask.lenses.io/u/stheppi)\
**Post date:** [April 17, 2023, 7:24am UTC](https://ask.lenses.io/t/kcql-function-doesnt-work-in-influx-db-sink-connector/43/2 "2023-04-17T07:24:31Z")

</div>

Lenses KCQL is a SQL-like DSL to configure Kafka connectors. However, it is not full-blown SQL. In this case, you use a function: COALESCE, but functions are not supported.

KCQL supports only field selection and metadata settings:

```auto
INSERT INTO <your-measure>
SELECT FIELD, ...
FROM kafka_topic_name
[WITHTIMESTAMP FIELD|sys_time]
[WITHTAG(FIELD|(constant_key=constant_value)]

```

To work around your challenge, for example, you can rely on [Kafka Connect SMT](https://docs.confluent.io/platform/current/connect/transforms/overview.html), but you’ll have to write one since there is no support in the provided transformations. Alternatively, you can transform your topic data stream and use, for example, Lenses SQL processors to prepare the payload before you sink it. You can run a SQL processor:

```auto
INSERT INTO orders_fixed
SELECT STREAM 
       id
      , name
      , order_code
      , COALESCE(order_type, ‘’) AS order_type 
FROM orders

```

And then your connector config KCQL would be:

```auto
NSERT INTO orders_model1 SELECT * FROM orders WITHTIMESTAMP sys_time()

```