Fred Teunissen created FLINK-40082:
--------------------------------------
Summary: InvalidProgramException when performing LEFT JOIN with a
lookup table that has a computed field for Watermark extraction based on
metadata field (HEADERS)
Key: FLINK-40082
URL: https://issues.apache.org/jira/browse/FLINK-40082
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.2.1, 2.3.0, 2.2.0
Reporter: Fred Teunissen
Attachments: invalid-program-exception-reproduce-on-docker.tar.gz
On Flink 2.2.1 and 2.3.0 the following issue occurs when performing a LEFT JOIN
with a lookup table that is using a *{*}computed field{*}* to extract the
Watermark from a metadata field (HEADERS). The Job aborts with an
InvalidProgramException.
On Flink 2.1.2 the same job runs without any issues.
{noformat}
java.lang.RuntimeException: Could not instantiate generated class
'WatermarkGenerator$0'
at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:74)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedWatermarkGeneratorSupplier.createWatermarkGenerator(GeneratedWatermarkGeneratorSupplier.java:71)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.api.operators.source.ProgressiveTimestampsAndWatermarks.createMainOutput(ProgressiveTimestampsAndWatermarks.java:128)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.api.operators.SourceOperator.initializeMainOutput(SourceOperator.java:554)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.api.operators.SourceOperator.emitNextNotReading(SourceOperator.java:526)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:502)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:646)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:988)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:925)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:973)
~[flink-dist-2.2.1.jar:2.2.1]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:955)
~[flink-dist-2.2.1.jar:2.2.1]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:760)
~[flink-dist-2.2.1.jar:2.2.1]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:569)
~[flink-dist-2.2.1.jar:2.2.1]
at java.base/java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: org.apache.flink.util.FlinkRuntimeException:
org.apache.flink.api.common.InvalidProgramException: Table program cannot be
compiled. This is a bug. Please file an issue.
at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:94)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
~[flink-table-runtime-2.2.1.jar:2.2.1]
... 16 more
Caused by:
org.apache.flink.shaded.guava33.com.google.common.util.concurrent.UncheckedExecutionException:
org.apache.flink.api.common.InvalidProgramException: Table program cannot be
compiled. This is a bug. Please file an issue.
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2086)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
~[flink-table-runtime-2.2.1.jar:2.2.1]
... 16 more
Caused by: org.apache.flink.api.common.InvalidProgramException: Table program
cannot be compiled. This is a bug. Please file an issue.
at
org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:107)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4903)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3574)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2316)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2190)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2080)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
~[flink-table-runtime-2.2.1.jar:2.2.1]
... 16 more
Caused by: org.codehaus.commons.compiler.CompileException: Line 136, Column 86:
Thrown exception of type "java.io.UnsupportedEncodingException" is neither
caught by a "try...catch" block nor declared in the "throws" clause of the
declaring function
at org.codehaus.janino.UnitCompiler.compileError(UnitCompiler.java:13080)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler.checkThrownException(UnitCompiler.java:10129)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.invokeConstructor(UnitCompiler.java:8354)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5640)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$10000(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$16.visitNewClassInstance(UnitCompiler.java:4778)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$16.visitNewClassInstance(UnitCompiler.java:4742)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$NewClassInstance.accept(Java.java:5587)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4742)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGetValue(UnitCompiler.java:5885)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGet2(UnitCompiler.java:5391)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$9300(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4766)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$16.visitMethodInvocation(UnitCompiler.java:4742)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$MethodInvocation.accept(Java.java:5470)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGet(UnitCompiler.java:4742)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileGetValue(UnitCompiler.java:5885)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:4132)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$6300(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$13.visitAssignment(UnitCompiler.java:4088)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$13.visitAssignment(UnitCompiler.java:4071)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$Assignment.accept(Java.java:4864)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:4071)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:2524)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$1800(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$6.visitExpressionStatement(UnitCompiler.java:1581)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$6.visitExpressionStatement(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$ExpressionStatement.accept(Java.java:3209)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileStatements(UnitCompiler.java:1662)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:1647)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$1700(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$6.visitBlock(UnitCompiler.java:1580)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$6.visitBlock(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$Block.accept(Java.java:3115)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:2650)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$1900(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$6.visitIfStatement(UnitCompiler.java:1582)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler$6.visitIfStatement(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$IfStatement.accept(Java.java:3284)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:1576)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileStatements(UnitCompiler.java:1662)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:3665)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:3330)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1448)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler.compileDeclaredMethods(UnitCompiler.java:1421)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:830)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:443)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$400(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$3.visitPackageMemberClassDeclaration(UnitCompiler.java:423)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$3.visitPackageMemberClassDeclaration(UnitCompiler.java:419)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.Java$PackageMemberClassDeclaration.accept(Java.java:1688)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile(UnitCompiler.java:419)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compile2(UnitCompiler.java:393)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.access$000(UnitCompiler.java:237)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$2.visitCompilationUnit(UnitCompiler.java:364)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.UnitCompiler$2.visitCompilationUnit(UnitCompiler.java:362)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.Java$CompilationUnit.accept(Java.java:371)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.UnitCompiler.compileUnit(UnitCompiler.java:362)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:273)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.codehaus.janino.SimpleCompiler.compileToClassLoader(SimpleCompiler.java:526)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.SimpleCompiler.cook2(SimpleCompiler.java:250)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:229)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.janino.SimpleCompiler.cook(SimpleCompiler.java:219)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:82)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at org.codehaus.commons.compiler.Cookable.cook(Cookable.java:77)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.doCompile(CompileUtils.java:104)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.lambda$compile$0(CompileUtils.java:92)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache$1.load(LocalCache.java:4903)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3574)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2316)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2190)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2080)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache.get(LocalCache.java:4017)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.shaded.guava33.com.google.common.cache.LocalCache$LocalManualCache.get(LocalCache.java:4898)
~[flink-dist-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.CompileUtils.compile(CompileUtils.java:92)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.compile(GeneratedClass.java:101)
~[flink-table-runtime-2.2.1.jar:2.2.1]
at
org.apache.flink.table.runtime.generated.GeneratedClass.newInstance(GeneratedClass.java:68)
~[flink-table-runtime-2.2.1.jar:2.2.1]
... 16 more{noformat}
h2. Reproduce in docker environment.
To reproduce the issue, you can start the provided Docker Compose environment.
It will set up a Kafka cluster, Flink cluster and run a Flink SQL job that
contains the issue.
Start the Flink SQL job with the following command (from the folder containing
the `docker-compose.yaml` file):
{code:bash}
docker compose run --build --rm sqljob
{code}
You can see the error in the logs of the JobManager container (mapped on the
host to the `./logs/job/` folder).
The job will fail with an InvalidProgramException.
To run interactively with the SQL Client, use the following command to start
the environment without running the job:
{code:bash}
docker compose run --build --rm --interactive sqlclient
{code}
The `./sql/init.sql` file contains the SQL statements to create the source and
sink tables and will be automagically executed.
The `./sql/job.sql` file contains the SQL statements to execute the job that
performs the LEFT JOIN with the lookup table.
To cleanup the environment, use the following command:
{code:bash}
docker compose down
{code}
h2. Steps to Reproduce:
Create the input source table and output sink table:
{code:sql}
CREATE TABLE input (
id STRING,
ingestionTime TIMESTAMP(3),
WATERMARK FOR ingestionTime AS ingestionTime - INTERVAL '0.5' SECOND
) WITH (
'connector' = 'datagen',
'rows-per-second' = '500',
'number-of-rows' = '100000'
);
CREATE TABLE lookupTbl
(
id STRING NOT NULL,
WATERMARK FOR computedIngestTime AS computedIngestTime - INTERVAL '1.000'
SECOND,
headers MAP<STRING,BYTES> NOT NULL METADATA FROM 'headers',
computedIngestTime AS TO_TIMESTAMP(DECODE(headers['orig_ingest_ts'], 'UTF-8')),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
– connector options
'connector' = 'upsert-kafka',
'properties.group.id' = 'consumer-group-1',
'topic' = 'compacted_topic',
'properties.bootstrap.servers' = 'kafka:29092',
– format options for key options
'key.format' = 'raw',
– format options for value options
'value.format' = 'json'
);
CREATE TABLE blackhole_table
(
id STRING,
ingest_time TIMESTAMP(3)
) WITH (
'connector' = 'blackhole'
);
{code}
Execute the job that performs a CROSS JOIN with an inline VALUES clause:
{code:sql}
INSERT INTO blackhole_table
SELECT i.id AS id,
l.computedIngestTime AS ingest_time
FROM input AS i
LEFT JOIN lookupTbl FOR SYSTEM_TIME AS OF i.ingestionTime AS l
ON i.id = l.id;
{code}
h2. Expected Behavior
Job should run without any errors and produce output to the blackhole sink.
h2. Actual Behavior:
Job fails with an InvalidProgramException when trying to compile the generated
code for the WatermarkGenerator class, due to an uncaught
UnsupportedEncodingException in the generated code.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
