debezium avro format cannot include header

[gaozuo]

Describe the bug

I need a data change time, but the fields in the reference document report an error. Is there any way to generate a data change time? My raw data is already in debezium avro format

Error message/log

CREATE source cdc_shareholder
include timestamp as timestamp 
WITH (
  connector = 'kafka', 
  properties.bootstrap.server = 'kafka-stream-kafka-bootstrap.kafka-cluster:9092', 
  scan.startup.mode = 'earliest', 
  topic = 'HR.HROUTPUT.SHAREHOLDER'
)
FORMAT DEBEZIUM ENCODE AVRO (
  schema.registry = 'http://kafka-stream-registry-cp-schema-registry.kafka-cluster:8081'
);


org.jkiss.dbeaver.model.sql.DBSQLException: SQL 错误 [XX000]: ERROR: Failed to execute the statement

Caused by:
  Protocol error: FORMAT DEBEZIUM forbids additional columns, but got ["timestamp"]


	at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCStatementImpl.executeStatement(JDBCStatementImpl.java:133)

	at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.executeStatement(SQLQueryJob.java:614)

	at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.lambda$2(SQLQueryJob.java:505)

	at org.jkiss.dbeaver.model.exec.DBExecUtils.tryExecuteRecover(DBExecUtils.java:194)

	at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.executeSingleQuery(SQLQueryJob.java:524)

	at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.extractData(SQLQueryJob.java:976)

	at org.jkiss.dbeaver.ui.editors.sql.SQLEditor$QueryResultsContainer.readData(SQLEditor.java:4155)

	at org.jkiss.dbeaver.ui.controls.resultset.ResultSetJobDataRead.lambda$0(ResultSetJobDataRead.java:123)

	at org.jkiss.dbeaver.model.exec.DBExecUtils.tryExecuteRecover(DBExecUtils.java:194)

	at org.jkiss.dbeaver.ui.controls.resultset.ResultSetJobDataRead.run(ResultSetJobDataRead.java:121)

	at org.jkiss.dbeaver.ui.controls.resultset.ResultSetViewer$ResultSetDataPumpJob.run(ResultSetViewer.java:5152)

	at org.jkiss.dbeaver.model.runtime.AbstractJob.run(AbstractJob.java:115)

	at org.eclipse.core.internal.jobs.Worker.run(Worker.java:63)

Caused by: org.postgresql.util.PSQLException: ERROR: Failed to execute the statement

Caused by:
  Protocol error: FORMAT DEBEZIUM forbids additional columns, but got ["timestamp"]


	at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2725)

	at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2412)

	at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:371)

	at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:502)

	at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:419)

	at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:341)

	at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:326)

	at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:302)

	at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:297)

	at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCStatementImpl.execute(JDBCStatementImpl.java:330)

	at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCStatementImpl.executeStatement(JDBCStatementImpl.java:131)

	... 12 more

To Reproduce

No response

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response

[tabVersion]

can you help ref the doc page here?

& from the SQL above, I don't see you are using the message header but the timestamp. Can you clarify the requirement?

[gaozuo]

Here I need to get the change time of the data. Currently I want to quote the timestamp of kafka, but according to the example in the document, it cannot.
see this page: https://docs.risingwave.com/docs/current/include-clause/#examples

[tabVersion]

Here I need to get the change time of the data. Currently I want to quote the timestamp of kafka, but according to the example in the document, it cannot. see this page: docs.risingwave.com/docs/current/include-clause#examples

I get it. But the kafka message time can differ from the debezium message's event time.

From the impl side, the additional column timestamp is not supported for format debezium. We are still not sure what the column stands for in such scene, ie. kafka message timestamp or debezium's event time.