https://flink.apache.org/ logo
Join Slack
Powered by
I'm getting the following exception ```org.apache...
# troubleshooting
c
I'm getting the following exception
Copy code
org.apache.flink.table.api.ValidationException: Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
Cause: Different number of columns.

Query schema: [id: STRING NOT NULL, enrichment: STRING NOT NULL, ts: TIMESTAMP(3) *ROWTIME*]
Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
Based on https://cwiki.apache.org/confluence/display/FLINK/FLIP-66%3A+Support+time+attribute+in+SQL+DDL I think I need to set a watermark strategy and perhaps rowtime field for my table job. Can I set this on the Table schema below? Is there a flink doc that shows this or would someone please kindly provide the syntax for doing this with the Table API?
Copy code
Schema.newBuilder()
                .column("id", DataTypes.STRING().notNull())
                .column("status", DataTypes.STRING().notNull())
                .column("enrichment", DataTypes.STRING().notNull())
                .column("ts",DataTypes.TIMESTAMP(3).bridgedTo(java.sql.Timestamp.class))
                .primaryKey("id")
                .watermark("ts","ts")
                .build();
m
Isn’t this because you don’t have a “status” field in your query schema?
c
Hmm I'm suprised by that because I was attempting to join records
m
But are you selecting a “status” field as a result of your join?
c
Yeah let me share what I'm doing. I will just push it to github. One moment please.
Here are the schemas
Looks like I was executing the wrong query... 😞 . Well I will fix that.
m
We’ve all been there 😅
c
I fixed that but now my join gets
Copy code
org.apache.flink.table.api.ValidationException: Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
        Cause: Different number of columns.

        Query schema: [id: STRING NOT NULL, status: STRING, ts: TIMESTAMP(3) *ROWTIME*, id0: STRING NOT NULL, enrichment: STRING NOT NULL, ts0: TIMESTAMP(3) *ROWTIME*]
        Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
Also you can run it locally by checking out the branch. It looks like the join returning both rows. Do I just need to select the fields that I want and expect the join to work?
IE SELECT id, status, enrichment, ts explicitly? Or do I need to set tat ROWTIME attribute in my sink schema or am I doing something terrible wrong?
I tried changing the select like ^^ 
public static final String JOIN_QUERY_2 = "INSERT INTO "+ENR_DST_TABLE_PATH+" SELECT id0,status,ts,enrichment FROM "+SRC_TABLE_PATH+ " JOIN "+ENR_SRC_TABLE_PATH+" FOR SYSTEM_TIME AS OF "+SRC_TABLE_PATH+".ts"+" ON "+SRC_TABLE_PATH+".id ="+ENR_SRC_TABLE_PATH+".id";
But that didn't seem to change the query schema to match the sink schema like I had hoped. Is there something obviously wrong with my approach?
Copy code
org.apache.flink.table.api.ValidationException: Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
Cause: Different number of columns.

Query schema: [id: STRING NOT NULL, status: STRING, ts: TIMESTAMP(3) *ROWTIME*, id0: STRING NOT NULL, enrichment: STRING NOT NULL, ts0: TIMESTAMP(3) *ROWTIME*]
Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
My expectation is that the query schema would match the sink schema based on the select. Am I nuts or am I doing something wrong?
I'll fiddle with it some more and see if I can get some different behavior. But I'm confused now.
I reshaped my query giving aliases to the tables for the join. So I resolved the selcting problem. But I'm still having the 
*ROWTIME
that came as the hypothesis of the original post
Copy code
org.apache.flink.table.api.ValidationException: Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
        Cause: Incompatible types for sink column 'enrichment' at position 2.

        Query schema: [id: STRING NOT NULL, status: STRING, ts: TIMESTAMP(3) *ROWTIME*, enrichment: STRING NOT NULL]
        Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
m
Isn't it complaining that 
enrichment
should also support 
NOT NULL
for the sink?
c
@Martijn Visser the schema does show NOT NULL in it's definition
m
Copy code
Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
That doesn't, right?
c
m
I can have a look tomorrow, when I'm back on a laptop 🙂
c
Thank you kindly. I will keep tooling with it and keep pushing to my branch, hoping I can resolve it so you don't need to.
I pushed more changes to the branch but I have the issue. I am assuming that it's related to 
*ROWTIME
missing in the output schema. Otherwise if you're correct then it's ignoring the output schema fields that I have shown I'm explicitly setting to null. It's a loss to me because I'm writing my query like the example. https://github.com/ververica/flink-sql-cookbook/blob/main/joins/03_kafka_join/03_kafka_join.md
It doesn't seem reasonable calling 
executeSQL
with`.print()` would cause this. But still not sure why I cannot execute the query without 
print()
https://github.com/apple-corps/recipes/blob/14d0a7d51ff9cddbd9550d7efe3b0edee85948[…]/table-example/src/main/java/com/immerok/cookbook/TableJob.java
m
But still not sure why I cannot execute the query without 
print()
You should, https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/insert/ and it's also used numerous times in Flink codebase itself
You could first try to just to just do the SELECT and print that to screen (I can't get your code to compile at the moment). If that works, we've narrowed it down to a mismatch between the result of the SELECT and the schema you're trying to sink to
c
Not sure why it wouldn't compile. Is working fine for me.
Copy code
recipes/table-example:table-example -$ mvn install
[INFO] Scanning for projects...
[INFO] 
[INFO] -----------------< com.immerok:cookbook-table-example >-----------------
[INFO] Building Flink Cookbook Recipe - Table Example 0.1
[INFO] --------------------------------[ jar ]---------------------------------
[INFO] 
[INFO] --- maven-resources-plugin:2.6:resources (default-resources) @ cookbook-table-example ---
[INFO] Using 'UTF-8' encoding to copy filtered resources.
[INFO] Copying 1 resource
[INFO] 
[INFO] --- maven-compiler-plugin:3.10.1:compile (default-compile) @ cookbook-table-example ---
[INFO] Changes detected - recompiling the module!
[INFO] Compiling 4 source files to /home/cwilliams/code/recipes/table-example/target/classes
[INFO] 
[INFO] --- maven-resources-plugin:2.6:testResources (default-testResources) @ cookbook-table-example ---
[INFO] Using 'UTF-8' encoding to copy filtered resources.
[INFO] Copying 1 resource
[INFO] 
[INFO] --- maven-compiler-plugin:3.10.1:testCompile (default-testCompile) @ cookbook-table-example ---
[INFO] Changes detected - recompiling the module!
[INFO] Compiling 5 source files to /home/cwilliams/code/recipes/table-example/target/test-classes
[INFO] /home/cwilliams/code/recipes/table-example/src/test/java/com/immerok/cookbook/CookbookKafkaCluster.java: Some input files use unchecked or unsafe operations.
[INFO] /home/cwilliams/code/recipes/table-example/src/test/java/com/immerok/cookbook/CookbookKafkaCluster.java: Recompile with -Xlint:unchecked for details.
[INFO] 
[INFO] --- maven-surefire-plugin:2.22.2:test (default-test) @ cookbook-table-example ---
[INFO] 
[INFO] -------------------------------------------------------
[INFO]  T E S T S
[INFO] -------------------------------------------------------
[INFO] Running TableJobTest
06:02:53,475 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.cpu.cores required for local execution is not set, setting it to the maximal possible value.
06:02:53,477 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.task.heap.size required for local execution is not set, setting it to the maximal possible value.
06:02:53,477 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.task.off-heap.size required for local execution is not set, setting it to the maximal possible value.
06:02:53,478 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.network.min required for local execution is not set, setting it to its default value 64 mb.
06:02:53,478 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils [] - The configuration option taskmanager.memory.network.max required for local execution is not set, setting it to its default value 64 mb.
06:02:53,485 INFO  org.apache.flink.runtime.minicluster.MiniCluster             [] - Starting Flink Mini Cluster
06:02:53,489 INFO  org.apache.flink.runtime.minicluster.MiniCluster             [] - Starting Metrics Registry
06:02:53,513 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Stop job leader service.
06:02:55,918 INFO  org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager [] - Shutting down TaskExecutorLocalStateStoresManager.
06:02:55,919 INFO  org.apache.flink.runtime.io.disk.FileChannelManagerImpl      [] - FileChannelManager removed spill file directory /tmp/junit2307088322128074339/junit17902656541890378625/flink-io-d8a12e15-1482-48a8-9664-0f474bd4e400
06:02:55,919 INFO  org.apache.flink.runtime.io.network.NettyShuffleEnvironment  [] - Shutting down the network environment and its components.
06:02:55,920 INFO  org.apache.flink.runtime.io.disk.FileChannelManagerImpl      [] - FileChannelManager removed spill file directory /tmp/junit2307088322128074339/junit17902656541890378625/flink-netty-shuffle-69ee3243-23a5-4dd5-afac-ced686ee553f
06:02:55,920 INFO  org.apache.flink.runtime.taskexecutor.KvStateService         [] - Shutting down the kvState service and its components.
06:02:55,921 INFO  org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Stop job leader service.
06:02:55,921 INFO  org.apache.flink.runtime.filecache.FileCache                 [] - removed file cache directory /tmp/junit2307088322128074339/junit17902656541890378625/flink-dist-cache-0552221a-e96b-47b2-b00e-ab3b3dffd5be
06:02:55,921 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Stopped TaskExecutor <akka://flink/user/rpc/taskmanager_0>.
06:02:55,923 INFO  org.apache.flink.runtime.resourcemanager.StandaloneResourceManager [] - Shut down cluster because application is in CANCELED, diagnostics DispatcherResourceManagerComponent has been closed..
06:02:55,923 INFO  org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent [] - Closing components.
06:02:55,924 INFO  org.apache.flink.runtime.dispatcher.runner.SessionDispatcherLeaderProcess [] - Stopping SessionDispatcherLeaderProcess.
06:02:55,924 INFO  org.apache.flink.runtime.dispatcher.StandaloneDispatcher     [] - Stopping dispatcher <akka://flink/user/rpc/dispatcher_2>.
06:02:55,924 INFO  org.apache.flink.runtime.dispatcher.StandaloneDispatcher     [] - Stopping all currently running jobs of dispatcher <akka://flink/user/rpc/dispatcher_2>.
06:02:55,924 INFO  org.apache.flink.runtime.resourcemanager.ResourceManagerServiceImpl [] - Stopping resource manager service.
06:02:55,924 INFO  org.apache.flink.runtime.dispatcher.StandaloneDispatcher     [] - Stopped dispatcher <akka://flink/user/rpc/dispatcher_2>.
06:02:55,924 INFO  org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Closing the slot manager.
06:02:55,925 INFO  org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Suspending the slot manager.
06:02:55,926 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService             [] - Stopping Akka RPC service.
06:02:55,950 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService             [] - Stopping Akka RPC service.
06:02:55,950 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService             [] - Stopped Akka RPC service.
06:02:55,955 INFO  org.apache.flink.runtime.blob.PermanentBlobCache             [] - Shutting down BLOB cache
06:02:55,956 INFO  org.apache.flink.runtime.blob.TransientBlobCache             [] - Shutting down BLOB cache
06:02:55,956 INFO  org.apache.flink.runtime.blob.BlobServer                     [] - Stopped BLOB server at 0.0.0.0:34419
06:02:55,956 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService             [] - Stopped Akka RPC service.
[ERROR] Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 2.639 s <<< FAILURE! - in TableJobTest
[ERROR] testExecuteSQL  Time elapsed: 1.814 s  <<< ERROR!
org.apache.flink.table.api.ValidationException: 
Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
Cause: Incompatible types for sink column 'enrichment' at position 2.

Query schema: [id: STRING NOT NULL, status: STRING NOT NULL, ts: TIMESTAMP(3) *ROWTIME*, enrichment: STRING NOT NULL]
Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
        at TableJobTest.testExecuteSQL(TableJobTest.java:83)

[INFO] 
[INFO] Results:
[INFO] 
[ERROR] Errors: 
[ERROR]   TableJobTest.testExecuteSQL:83 » Validation Column types of query result and s...
[INFO] 
[ERROR] Tests run: 1, Failures: 0, Errors: 1, Skipped: 0
[INFO] 
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  4.531 s
[INFO] Finished at: 2023-01-18T06:02:56-08:00
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-surefire-plugin:2.22.2:test (default-test) on project cookbook-table-example: There are test failures.
[ERROR] 
[ERROR] Please refer to /home/cwilliams/code/recipes/table-example/target/surefire-reports for the individual test results.
[ERROR] Please refer to dump files (if any exist) [date].dump, [date]-jvmRun[N].dump and [date].dumpstream.
[ERROR] -> [Help 1]
[ERROR] 
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR] 
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] <http://cwiki.apache.org/confluence/display/MAVEN/MojoFailureException>
Copy code
table-example 130$ git log -1 
commit e4b0011b6842109d28688560d1e4fb82581e3292 (HEAD -> table-example, origin/table-example)
Author: Colin Williams <Colin Williams>
Date:   Tue Jan 17 15:32:43 2023 -0800

    iterated between kafka and kafka-upsert. issue still exists
I'll give the select a shot.
@Martijn Visser so the select works. But I'm at a loss why my Sink Schema doesn't match . We see the error above again when I run with the INSERT statement
Copy code
Query schema: [id: STRING NOT NULL, status: STRING NOT NULL, ts: TIMESTAMP(3) *ROWTIME*, enrichment: STRING NOT NULL]
Sink schema:  [id: STRING, status: STRING, enrichment: STRING, ts: TIMESTAMP(3)]
It looks like it's either missing 
NOT NULL
which is set in the "Sink Schema" for all the fields, or that it's the issue with TIMESTAMP(3) missing 
*ROWTIME*
@Martijn Visser we're at the same place I was at yesterday.
@Martijn Visser I show how I'm setting the schema for the table descriptor, creating the table, providing the query ^^
I'm able to build and run from linux and on the mac from the CLI and IDEA.
Copy code
git status
On branch table-example
Your branch is up to date with 'origin/table-example'.

nothing to commit, working tree clean
I pushed to the repo and showed the build above. Not sure what build error you see. Would you share it?
m
Not sure what build error you see. Would you share it?
Local Maven artifact was corrupt because I've been building new Flink Shaded versions locally 🙈
It's the 
enrichment
definition that causes issues
Column types of query result and sink for 'temporaryCatalog.default.enrichmentOut' do not match.
Cause: Incompatible types for sink column 'enrichment' at position 2.
c
m
Yeah same for me
Let's file a Jira ticket and ask some of the maintainers to take a look
c
Should I file the ticket?
m
That would be great yes
c
Is the component Table SQL / API?
m
Yes
c
Perhaps I'll try to follow it through the debugger myself.
I was able to make the following sql ddls to create a running job based on the schemas previously defined. https://github.com/apple-corps/recipes/blob/flink-sql/table-example/src/main/java/com/immerok/cookbook/schemas/ddl.txt I tried comparing against generated table ddls https://github.com/apple-corps/recipes/blob/flink-sql/table-example/src/main/java/com/immerok/cookbook/schemas/table-ddl.txt . I don't see anything that would cause the table job not to run with the exception we receive, but I could be wrong.
7 Views
Open in Slack
PreviousNext
Powered by