SQL PTransform Update. #112
Conversation
|
|
||
| @Override | ||
| public PDone expand(PBegin input) { | ||
| return PDone.in(input.getPipeline()); |
There was a problem hiding this comment.
Shall we log the statements and catalogs here? Seems better if we log them somewhere for debugging purpose.
Btw, there is an overridable "PTransform.validate(PipelineOptions)" method that you can use to do validations. I think we can double check things like statements are not empty and catalog is valid.
There was a problem hiding this comment.
We are logging the full statements in the translator. I changed that logging to info and added a debug level logging here. Good point about validation. I added an empty statement check there.
| StreamStatementSet ss = tEnv.createStatementSet(); | ||
| for (String statement : sqlTransform.getStatements()) { | ||
| combinedStatements.add(statement); | ||
| if (statement.substring(0, INSERT_INTO.length()).toUpperCase().startsWith(INSERT_INTO)) { |
There was a problem hiding this comment.
nit: maybe move this line of logic into a static method like isInsertIntoStatement()? A bit easier to read.
| } | ||
| // Now attach everything to StreamExecutionEnv. | ||
| ss.attachAsDataStream(); | ||
| LOG.debug("Executing SQL statements:\n {}", combinedStatements); |
There was a problem hiding this comment.
I would just make it info log so it's clear :)
| public class StatementOnlySqlTransform extends PTransform<PBegin, PDone> { | ||
|
|
||
| private final List<String> statements; | ||
| private final Map<String, SerializableCatalog> catalogs; |
There was a problem hiding this comment.
Just curious: we can support multiple catalogs here? Curious what's the use case would look like.
There was a problem hiding this comment.
Yes, sometimes the datasets may come from different external storage systems. For example, we may have a job reading from Hive, MySql and Kafka at the same time. In this case, there might be three catalogs each for one of the external system.
| ss.addInsertSql(statement); | ||
| } else { | ||
| // Not an insert into statement. Treat it as a DDL. | ||
| tEnv.executeSql(statement); |
There was a problem hiding this comment.
Can there be valid normal query statements without INSERT INTO? I am not very sure how FLink SQL statements look like today. Maybe everything except DDL has INSERT INTO in the beginning?
There was a problem hiding this comment.
A query logic in SQL is always a SELECT statement. In SQL the query logic has to be represented by some entity, and such entity would either be a Table or a View. So any SELECT statement will either be a part of CREATE TABLE statement or CREATE VIEW statement. INSERT INTO is effectively binding a SELECT statement with an existing table instead of creating a new one.
From execution perspective, only INSERT INTO will trigger an action to run the query. Without an INSERT INTO statement, all the queries will only generate temporary views or tables. That is why we can call tEnv.executeSql() for all the statements other than INSERT INTO, because they will not really trigger an execution. And we have to append all the INSERT INTO statement in a StatementSet so that multiple INSERT INTO can be executed in the same Flink job.
becketqin
force-pushed
the
sql_ptransform_update
branch
from
d30d083 to
5ea8e10
Compare
There was a problem hiding this comment.
Are these statements still valid ?
1 - https://jarvis.corp.linkedin.com/codesearch/result?name=MultiOutputSqlTransform.java&path=beam%2Frunners%2Fflink%2F1.15%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fflink%2Ftransform%2Fsql&reponame=linkedin%2Fbeam#42
And here
2 - https://jarvis.corp.linkedin.com/codesearch/result/?path=beam%2Frunners%2Fflink%2F1.15%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fflink%2Ftransform%2Fsql&reponame=linkedin%2Fbeam&name=SingleOutputSqlTransform.java#40
becketqin
force-pushed
the
sql_ptransform_update
branch
from
5ea8e10 to
b645f96
Compare
| @@ -114,6 +114,16 @@ public static <T> SingleOutputSqlTransform<T> of(Class<T> outputClass) { | |||
| return new SingleOutputSqlTransform<>(of(Integer.class, outputClass)); | |||
| } | |||
|
|
|||
| /** | |||
| * Create a {@link StatementOnlySqlTransform} which takes a full script of SQL statements and | |||
| * execute them. The statements must have at least one <code>INSERT INTO</code> statement. | |||
There was a problem hiding this comment.
@becketqin Sorry for commenting after the PR is merged. Couple of clarifying questions.
- What if the SQL statements had a
INSERT OVERWRITEinstead ofINSERT INTO? Is that not a validStatementOnlySqlTransform? - Similarly, how about having a
CREATE TABLE AS SELECTwith outINSERT INTO, should we support that as well?
There was a problem hiding this comment.
That is a good point. We should support both. I'll have a follow up patch.
There was a problem hiding this comment.
I created the PR. However, currenlty we cannot support CTAS because the OSS Flink StreamStatementSet.attachAsDataStream() does not support that yet. We will need to make change to the OSS Flink first.
The existing SqlPTransform has a few limitations by design:
This patch removes these two restrictions.