You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by "Aireed (via GitHub)" <gi...@apache.org> on 2026/10/20 08:54:23 UTC
[I] [Bug]: miss projection and watermark in LogDynamicSource::copy method [amoro]
Aireed opened a new issue, #4455:
URL: https://github.com/apache/amoro/issues/4455
### What happened?
1. define one amoro mixed_hive table with logstore enabled,
<img width="891" height="416" alt="Image" src="https://github.com/user-attachments/assets/e5c0c384-45b6-4fa4-b8d5-7c814f31aba0" />
2. Flink SQL: use `create table like ` to define watermark (only use part columns of table schema)
> create table tmp_xzl_ai_beh (
market_id AS COALESCE(JSON_VALUE(btn_title, '$.market_id'), 'unknown'),
tab_name AS COALESCE(JSON_VALUE(btn_title, '$.tab_name'), 'unknown'),
`recvtime_ts` AS CAST(recvtime AS TIMESTAMP(3)),
WATERMARK FOR recvtime_ts AS recvtime_ts - INTERVAL '5' SECOND
) like `amoro_catalog`.`amoro_v7`.`ht_pk_logstore_1`;
3. create a view from logstore
> CREATE view evt_view AS
SELECT
deviceid,
tab_name,
recvtime_ts,
market_id,
stock_code
FROM tmp_xzl_ai_beh/*+ OPTIONS('streaming' = 'true','arctic.read.mode' = 'log','scan.startup.mode'='latest') */;
4、insert into xxx select * from evt_view, and the exception thrown
> 2026-10-15 17:46:36
java.lang.ClassCastException: org.apache.flink.table.data.binary.BinaryStringData cannot be cast to org.apache.flink.table.data.TimestampData
at org.apache.flink.table.data.GenericRowData.getTimestamp(GenericRowData.java:179)
at org.apache.flink.table.data.RowData.lambda$createFieldGetter$39385f9c$1(RowData.java:260)
at org.apache.flink.table.data.RowData.lambda$createFieldGetter$25774257$1(RowData.java:296)
at org.apache.flink.table.runtime.typeutils.RowDataSerializer.toBinaryRow(RowDataSerializer.java:207)
at org.apache.flink.table.runtime.typeutils.RowDataSerializer.serialize(RowDataSerializer.java:103)
at org.apache.flink.table.runtime.typeutils.RowDataSerializer.serialize(RowDataSerializer.java:48)
at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.serialize(StreamElementSerializer.java:165)
at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.serialize(StreamElementSerializer.java:43)
at org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:54)
at org.apache.flink.runtime.io.network.api.writer.RecordWriter.serializeRecord(RecordWriter.java:141)
at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:107)
at org.apache.flink.runtime.io.network.api.writer.ChannelSelectorRecordWriter.emit(ChannelSelectorRecordWriter.java:55)
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:134)
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collectAndCheckIfChained(RecordWriterOutput.java:114)
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:95)
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:48)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:59)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:31)
at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:309)
at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
at org.apache.amoro.flink.read.source.log.kafka.LogKafkaRecordEmitter.emitRecord(LogKafkaRecordEmitter.java:41)
at org.apache.amoro.flink.read.source.log.kafka.LogKafkaRecordEmitter.emitRecord(LogKafkaRecordEmitter.java:28)
at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:160)
at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:419)
at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.lang.Thread.run(Thread.java:748)
### Affects Versions
master
### What table formats are you seeing the problem on?
Mixed-Hive
### What engines are you seeing the problem on?
Flink
### How to reproduce
_No response_
### Relevant log output
```shell
```
### Anything else
_No response_
### Are you willing to submit a PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's Code of Conduct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@amoro.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: miss projection and watermark in LogDynamicSource::copy method [amoro]
Posted by "hutiefang76 (via GitHub)" <gi...@apache.org>.
hutiefang76 commented on issue #4455:
URL: https://github.com/apache/amoro/issues/4455#issuecomment-6327829539
Hi @Aireed, have you started working on a patch? I can pick up a focused fix for `LogDynamicSource.copy()` to preserve projection and watermark state, with regression tests for both. If you already have a patch underway, I can help review or test it instead.
<sub>AI-assisted.</sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org