getting 'Insert column name does not exist in target table' from running insert query using kafka connector with json table definition

[baharhmohammadi]

Hi 👋
I'm trying to insert to a kafka topic's table by running an insert query. We have set up Kafka connector in our Trino staging cluster successfully and I have added a json table definition to enable inserting directly to the kafka topic's table. When I run the insert query I get the following error. I appreciate any help regarding this
Query failed (#20220427_223253_00282_52nkk): Insert column name does not exist in target table: id io.trino.spi.TrinoException: Insert column name does not exist in target table: id
and the insert query is;
Insert into kafka_regional.default."_test-kafka-topic" (id, origin, action, payload) VALUES ('test_source', 'test_origin', 'create', 'test')
The json definition table I added is:
{ "tableName": "_test-kafka-topic", "schemaName": "default", "topicName": "default._test-kafka-topic, "message": { "dataFormat": "json", "fields": [ { "name": "id", "type": "VARCHAR", "mapping": "id" }, { "name": "origin", "type": "VARCHAR", "mapping": "origin" }, { "name": "action", "type": "VARCHAR", "mapping": "action" }, { "name": "payload", "type": "VARCHAR", "mapping": "payload" } ] } }

This is the debug message I get when I run this insert query from CLI:
uery 20220429_220313_00231_sjsmc failed: Insert column name does not exist in target table: id io.trino.spi.TrinoException: Insert column name does not exist in target table: id at io.trino.sql.analyzer.SemanticExceptions.semanticException(SemanticExceptions.java:48) at io.trino.sql.analyzer.SemanticExceptions.semanticException(SemanticExceptions.java:43) at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitInsert(StatementAnalyzer.java:507) at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitInsert(StatementAnalyzer.java:404) at io.trino.sql.tree.Insert.accept(Insert.java:68) at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27) at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:421) at io.trino.sql.analyzer.StatementAnalyzer.analyze(StatementAnalyzer.java:384) at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:79) at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:71) at io.trino.execution.SqlQueryExecution.analyze(SqlQueryExecution.java:258) at io.trino.execution.SqlQueryExecution.<init>(SqlQueryExecution.java:182) at io.trino.execution.SqlQueryExecution$SqlQueryExecutionFactory.createQueryExecution(SqlQueryExecution.java:784) at io.trino.dispatcher.LocalDispatchQueryFactory.lambda$createDispatchQuery$0(LocalDispatchQueryFactory.java:135) at io.trino.$gen.Trino_371_001_1_g879661e____20220429_165101_2.call(Unknown Source) at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131) at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:74) at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82) 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)

[baharhmohammadi]

any updates on this? I'm trying to figure out where this error comes from? is it failing to access the file or access the table?

[baharhmohammadi]

we added some changes and now the insert query returns this error:
Query failed (#20220502_225802_00093_zxnic): Unexpected columns! expected: [KafkaColumnHandle{name=id, type=varchar, mapping=id, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}, KafkaColumnHandle{name=origin, type=varchar, mapping=origin, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}, KafkaColumnHandle{name=action, type=varchar, mapping=action, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}, KafkaColumnHandle{name=payload, type=varchar, mapping=payload, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}] got: [KafkaColumnHandle{name=id, type=varchar, mapping=id, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}, KafkaColumnHandle{name=origin, type=varchar, mapping=origin, dataFormat=null, formatHint=null, keyCodec=false, hidden=false, internal=false}, KafkaColumnHandle{name=action, type=varchar, mapping=action, dataFormat=null, formatHint= ...
Running describe kafka_regional.default."_test-kafka-topic" returns this schema:

and the insert query is the same one as above