+1 from my side.
Best,
Kurt
On Sun, May 17, 2020 at 12:41 PM godfrey he <godfre...@gmail.com> wrote:
hi everyone,
I would like to bring up another topic about the return value of
TableResult#collect() method.
Currently, the return type is `Iterator<Row>`, we meet some problems when
implementing FLINK-14807[1].
In current design, the sink operator has a buffer pool which buffers the
data from upstream,
and waits the client to consume the data. The client will pull the data
when `Iterator<Row>#next()` method is called.
If the client submits a select job, consumes a part of data and exits.
The
job will not be finished.
This will cause resource leak. We can't require the client must consume
all
data. for unbounded stream job, it's also impossible.
Currently, users can also cancel the job via
`TableResult.getJobClient().get().cancel()` method.
But this approach is not intuitive and convenient.
So, I want to change the return type from `Iterator<Row>` to
`CloseableRowIterator`,
the new method likes like:
public interface TableResult {
CloseableRowIterator collect();
}
public interface CloseableRowIterator extends Iterator<Row>,
AutoCloseable
{
}
Prefixing the name with "Closeable" is intended to remind the users that
this iterator should be closed,
users can conveniently use try-with-resources statement to close the
resources.
The resource leak problem is still there if users do not close the
iterator
or cancel the job through job client,
we just provide an easier way for users to avoid this.
I also notice that there is a `CloseableIterator` interface in
`org.apache.flink.util` package.
But I still tend to introduce `CloseableRowIterator`. My point of view
is:
1) `CloseableIterator` is in a util package, not a public interface.
2) `CloseableRowIterator` is more convenient, users do not need to define
generic type `<Row>`.
What do you think?
Best,
Godfrey
[1] https://issues.apache.org/jira/browse/FLINK-14807
Fabian Hueske <fhue...@gmail.com> 于2020年5月7日周四 下午3:59写道：
Thanks for the update Godfrey!
+1 to this approach.
Since there can be only one primary key, I'd also be fine to just use
`PRI` even if it is composite, but `PRI(f0, f5)` might be more
convenient
for users.
Thanks, Fabian
Am Do., 7. Mai 2020 um 09:31 Uhr schrieb godfrey he <
godfre...@gmail.com
:
Hi fabian,
Thanks for you suggestions.
Agree with you that `UNQ(f2, f3)` is more clear.
A table can have only ONE primary key,
this primary key can consist of single or multiple columns. [1]
if primary key consists of single column,
we can simply use `PRI` (or `PRI(xx)`) to represent it.
if primary key have multiple columns,
we should use `PRI(xx, yy, ...)` to represent it.
A table may have multiple unique keys,
each unique key can consist of single or multiple columns. [2]
if there is only one unique key and this unique key has only single
column,
we can simply use `UNQ` (or `UNQ(xx)`) to represent it.
otherwise, we should use `UNQ(xx, yy, ...)` to represent it.
(a corner case: two unique keys with same columns, like `UNQ(f2, f3)`,
`UNQ(f2, f3)`,
we can forbid this case or add a unique name for each key in the
future)
primary key and unique key with multiple columns example:
create table MyTable (
f0 BIGINT NOT NULL,
f1 ROW<q1 STRING, q2 TIMESTAMP(3)>,
f2 VARCHAR<256>,
f3 AS f0 + 1,
f4 TIMESTAMP(3) NOT NULL,
f5 BIGINT NOT NULL,
* PRIMARY KEY (f0, f5)*,
*UNIQUE (f3, f2)*,
WATERMARK f4 AS f4 - INTERVAL '3' SECOND
) with (...)
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| name | type
|
null | key | compute column | watermark
|
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f0 | BIGINT
|
false | PRI(f0, f5) | (NULL) | (NULL)
|
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f1 | ROW<q1 STRING, q2 TIMESTAMP(3)> | true | (NULL)
|
(NULL) | (NULL) |
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f2 | VARCHAR<256> | true
|
UNQ(f2, f3) | (NULL) | (NULL)
|
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f3 | BIGINT
|
false | UNQ(f2, f3) | f0 + 1 | (NULL)
|
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f4 | TIMESTAMP(3) |
false
| (NULL) | (NULL) | f4 - INTERVAL '3' SECOND |
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
| f5 | BIGINT
|
false | PRI(f0, f5) | (NULL) | (NULL)
|
+--------+------------------------------------------------------+-------+----------------+-----------------------+--------------------------------------+
"Regarding to the watermark on nested columns", that's a good approach
which can both support watermark on nested columns in the future and
keep
current table form.
[1] https://www.w3schools.com/sql/sql_primarykey.asp
[2] https://www.w3schools.com/sql/sql_unique.ASP
Best,
Godfrey
Fabian Hueske <fhue...@gmail.com> 于2020年5月7日周四 上午12:03写道：
Hi Godfrey,
This looks good to me.
One side note, indicating unique constraints with "UNQ" is probably
not
enough.
There might be multiple unique constraints and users would like to
know
which field combinations are unique.
So in your example above, "UNQ(f2, f3)" might be a better marker.
Just as a thought, if we would later add support for watermark on
nested
columns, we could add a row just for the nested field (in addition to
the
top-level field) like this:
+------------------------+---------------------------+-------+-----------+-------------+-----------------------------------------------------------+
| f4.nested.rowtime | TIMESTAMP(3) | false | (NULL) | (NULL)
| f4.nested.rowtime - INTERVAL '3' SECOND |
+------------------------+---------------------------+-------+-----------+-------------+-----------------------------------------------------------+
Thanks,
Fabian
Am Mi., 6. Mai 2020 um 17:51 Uhr schrieb godfrey he <
godfre...@gmail.com
:
Hi @fhue...@gmail.com @Timo Walther <twal...@apache.org> @Dawid
Wysakowicz <dwysakow...@apache.org>
What do you think we limit watermark must be defined on top-level
column ?
if we do that, we can add an expression column to represent
watermark
like compute column,
An example of all cases:
create table MyTable (
f0 BIGINT NOT NULL,
f1 ROW<q1 STRING, q2 TIMESTAMP(3)>,
f2 VARCHAR<256>,
f3 AS f0 + 1,
f4 TIMESTAMP(3) NOT NULL,
PRIMARY KEY (f0),
UNIQUE (f3, f2),
WATERMARK f4 AS f4 - INTERVAL '3' SECOND
) with (...)
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| name | type
| null | key | compute column | watermark
|
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| f0 | BIGINT
| false | PRI | (NULL) | (NULL)
|
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| f1 | ROW<q1 STRING, q2 TIMESTAMP(3)> | true | (NULL) |
(NULL) | (NULL) |
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| f2 | VARCHAR<256> | true
| UNQ | (NULL) | (NULL)
|
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| f3 | BIGINT
| false | UNQ | f0 + 1 | (NULL)
|
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
| f4 | TIMESTAMP(3) |
false | (NULL) | (NULL) | f4 - INTERVAL '3' SECOND |
+--------+------------------------------------------------------+-------+-----------+-----------------------+--------------------------------------+
WDYT ?
Best,
Godfrey
godfrey he <godfre...@gmail.com> 于2020年4月30日周四 下午11:57写道：
Hi Fabian,
the broken example is:
create table MyTable (
f0 BIGINT NOT NULL,
f1 ROW<q1 STRING, q2 TIMESTAMP(3)>,
f2 VARCHAR<256>,
f3 AS f0 + 1,
PRIMARY KEY (f0),
UNIQUE (f3, f2),
WATERMARK f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
) with (...)
name
type
key
compute column
watermark
f0
BIGINT NOT NULL
PRI
(NULL)
f1
ROW<`q1` STRING, `q2` TIMESTAMP(3)>
UNQ
(NULL)
f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
f2
VARCHAR<256>
(NULL)
NULL
f3
BIGINT NOT NULL
UNQ
f0 + 1
or we add a column to represent nullability.
name
type
null
key
compute column
watermark
f0
BIGINT
false
PRI
(NULL)
f1
ROW<`q1` STRING, `q2` TIMESTAMP(3)>
true
UNQ
(NULL)
f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
f2
VARCHAR<256>
true
(NULL)
NULL
f3
BIGINT
false
UNQ
f0 + 1
Hi Jark,
If we can limit watermark must be defined on top-level column,
this will become more simple.
Best,
Godfrey
Jark Wu <imj...@gmail.com> 于2020年4月30日周四 下午11:38写道：
Hi,
I'm in favor of Fabian's proposal.
First, watermark is not a column, but a metadata just like primary
key, so
shouldn't stand with columns.
Second, AFAIK, primary key can only be defined on top-level
columns.
Third, I think watermark can also follow primary key than only
allow
to
define on top-level columns.
I have to admit that in FLIP-66, watermark can define on nested
fields.
However, during implementation, I found that it's too complicated
to
do
that. We have refactor time-based physical nodes,
we have to use code generation to access event-time, we have to
refactor
FlinkTypeFactory to support a complex nested rowtime.
There is not much value of this feature, but introduce a lot of
complexity
in code base.
So I think we can force watermark define on top-level columns. If
user want
to define on nested columns,
he/she can use computed column to be a top-level column.
Best,
Jark
On Thu, 30 Apr 2020 at 17:55, Fabian Hueske <fhue...@gmail.com>
wrote:
Hi Godfrey,
The formatting of your example seems to be broken.
Could you send them again please?
Regarding your points
because watermark express can be a sub-column, just like
`f1.q2`
in above
example I give.
I would put the watermark information in the row of the
top-level
field and
indicate to which nested field the watermark refers.
Don't we have to solve the same issue for primary keys that are
defined on
a nested field?
A boolean flag can't represent such info. and I do know
whether
we will
support complex watermark expression involving multiple columns
in
the
future. such as: "WATERMARK FOR ts as ts + f1 + interval '1'
second"
You are right, a simple binary flag is definitely not sufficient
to
display
the watermark information.
I would put the expression string into the field, i.e., "ts +
f1 +
interval
'1' second"
For me the most important point of why to not show the watermark
as
a row
in the table is that it is not field that can be queried but
meta
information on an existing field.
For the user it is important to know that a certain field has a
watermark.
Otherwise, certain queries cannot be correctly specified.
Also there might be support for multiple watermarks that are
defined of
different fields at some point. Would those be printed in
multiple
rows?
Best,
Fabian
Am Do., 30. Apr. 2020 um 11:25 Uhr schrieb godfrey he <
godfre...@gmail.com
:
Hi Fabian, Aljoscha
Thanks for the feedback.
Agree with you that we can deal with primary key as you
mentioned.
now, the type column has contained the nullability attribute,
e.g. BIGINT
NOT NULL.
(I'm also ok that we use two columns to represent type just
like
mysql)
Why I treat `watermark` as a special row ?
because watermark express can be a sub-column, just like
`f1.q2`
in above
example I give.
A boolean flag can't represent such info. and I do know
whether
we will
support complex
watermark expression involving multiple columns in the future.
such as:
"WATERMARK FOR ts as ts + f1 + interval '1' second"
If we do not support complex watermark expression, we can add
a
watermark
column.
for example:
create table MyTable (
f0 BIGINT NOT NULL,
f1 ROW<q1 STRING, q2 TIMESTAMP(3)>,
f2 VARCHAR<256>,
f3 AS f0 + 1,
PRIMARY KEY (f0),
UNIQUE (f3, f2),
WATERMARK f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
) with (...)
name
type
key
compute column
watermark
f0
BIGINT NOT NULL
PRI
(NULL)
f1
ROW<`q1` STRING, `q2` TIMESTAMP(3)>
UNQ
(NULL)
f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
f2
VARCHAR<256>
(NULL)
NULL
f3
BIGINT NOT NULL
UNQ
f0 + 1
or we add a column to represent nullability.
name
type
null
key
compute column
watermark
f0
BIGINT
false
PRI
(NULL)
f1
ROW<`q1` STRING, `q2` TIMESTAMP(3)>
true
UNQ
(NULL)
f1.q2 AS (`f1.q2` - INTERVAL '3' SECOND)
f2
VARCHAR<256>
true
(NULL)
NULL
f3
BIGINT
false
UNQ
f0 + 1
Personally, I like the second one. (we need do some changes on
LogicalType
to get type name without nullability)
Best,
Godfrey
Aljoscha Krettek <aljos...@apache.org> 于2020年4月29日周三
下午5:47写道：
+1 I like the general idea of printing the results as a
table.
On the specifics I don't know enough but Fabians suggestions
seems to
make sense to me.
Aljoscha
On 29.04.20 10:56, Fabian Hueske wrote:
Hi Godfrey,
Thanks for starting this discussion!
In my mind, WATERMARK is a property (or constraint) of a
field, just
like
PRIMARY KEY.
Take this example from MySQL:
mysql> CREATE TABLE people (id INT NOT NULL, name
VARCHAR(128) NOT
NULL,
age INT, PRIMARY KEY (id));
Query OK, 0 rows affected (0.06 sec)
mysql> describe people;
+-------+--------------+------+-----+---------+-------+
| Field | Type | Null | Key | Default | Extra |
+-------+--------------+------+-----+---------+-------+
| id | int | NO | PRI | NULL | |
| name | varchar(128) | NO | | NULL | |
| age | int | YES | | NULL | |
+-------+--------------+------+-----+---------+-------+
3 rows in set (0.01 sec)
Here, PRIMARY KEY is marked in the Key column of the id
field.
We could do the same for watermarks by adding a Watermark
column.
Best, Fabian
Am Mi., 29. Apr. 2020 um 10:43 Uhr schrieb godfrey he <
godfre...@gmail.com>:
Hi everyone,
I would like to bring up a discussion about the result
type
of
describe
statement,
which is introduced in FLIP-84[1].
In previous version, we define the result type of
`describe`
statement
is a
single column as following
Statement
Result Schema
Result Value
Result Kind
Examples
DESCRIBE xx
field name: result
field type: VARCHAR(n)
(n is the max length of values)
describe the detail of an object
(single row)
SUCCESS_WITH_CONTENT
DESCRIBE table_name
for "describe table_name", the result value is the
`toString` value
of
`TableSchema`, which is an unstructured data.
It's hard to for user to use this info.
for example:
TableSchema schema = TableSchema.builder()
.field("f0", DataTypes.BIGINT())
.field("f1", DataTypes.ROW(
DataTypes.FIELD("q1", DataTypes.STRING()),
DataTypes.FIELD("q2", DataTypes.TIMESTAMP(3))))
.field("f2", DataTypes.STRING())
.field("f3", DataTypes.BIGINT(), "f0 + 1")
.watermark("f1.q2", WATERMARK_EXPRESSION,
WATERMARK_DATATYPE)
.build();
its `toString` value is:
root
|-- f0: BIGINT
|-- f1: ROW<`q1` STRING, `q2` TIMESTAMP(3)>
|-- f2: STRING
|-- f3: BIGINT AS f0 + 1
|-- WATERMARK FOR f1.q2 AS now()
For hive, MySQL, etc., the describe result is table form
including
field
names and field types.
which is more familiar with users.
TableSchema[2] has watermark expression and compute
column,
we
should
also
put them into the table:
for compute column, it's a column level, we add a new
column
named
`expr`.
for watermark expression, it's a table level, we add a
special row
named
`WATERMARK` to represent it.
The result will look like about above example:
name
type
expr
f0
BIGINT
(NULL)
f1
ROW<`q1` STRING, `q2` TIMESTAMP(3)>
(NULL)
f2
STRING
NULL
f3
BIGINT
f0 + 1
WATERMARK
(NULL)
f1.q2 AS now()
now there is a pr FLINK-17112 [3] to implement DESCRIBE
statement.
What do you think about this update?
Any feedback are welcome~
Best,
Godfrey
[1]
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=134745878
[2]
https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/api/TableSchema.java
[3] https://github.com/apache/flink/pull/11892
godfrey he <godfre...@gmail.com> 于2020年4月6日周一 下午10:38写道：
Hi Timo,
Sorry for the late reply, and thanks for your
correction.
I missed DQL for job submission scenario.
I'll fix the document right away.
Best,
Godfrey
Timo Walther <twal...@apache.org> 于2020年4月3日周五
下午9:53写道：
Hi Godfrey,
I'm sorry to jump in again but I still need to clarify
some things
around TableResult.
The FLIP says:
"For DML, this method returns TableResult until the job
is
submitted.
For other statements, TableResult is returned until the
execution
is
finished."
I thought we agreed on making every execution async?
This
also
means
returning a TableResult for DQLs even though the
execution
is not
done
yet. People need access to the JobClient also for batch
jobs in
order
to
cancel long lasting queries. If people want to wait for
the
completion
they can hook into JobClient or collect().
Can we rephrase this part to:
The FLIP says:
"For DML and DQL, this method returns TableResult once
the
job has
been
submitted. For DDL and DCL statements, TableResult is
returned
once
the
operation has finished."
Regards,
Timo
On 02.04.20 05:27, godfrey he wrote:
Hi Aljoscha, Dawid, Timo,
Thanks so much for the detailed explanation.
Agree with you that the multiline story is not
completed
now, and
we
can
keep discussion.
I will add current discussions and conclusions to the
FLIP.
Best,
Godfrey
Timo Walther <twal...@apache.org> 于2020年4月1日周三
下午11:27写道：
Hi Godfrey,
first of all, I agree with Dawid. The multiline story
is
not
completed
by this FLIP. It just verifies the big picture.
1. "control the execution logic through the proposed
method if
they
know
what the statements are"
This is a good point that also Fabian raised in the
linked
google
doc.
I
could also imagine to return a more complicated POJO
when
calling
`executeMultiSql()`.
The POJO would include some `getSqlProperties()` such
that a
platform
gets insights into the query before executing. We
could
also
trigger
the
execution more explicitly instead of hiding it behind
an
iterator.
2. "there are some special commands introduced in SQL
client"
For platforms and SQL Client specific commands, we
could
offer a
hook
to
the parser or a fallback parser in case the regular
table
environment
parser cannot deal with the statement.
However, all of that is future work and can be
discussed
in a
separate
FLIP.
3. +1 for the `Iterator` instead of `Iterable`.
4. "we should convert the checked exception to
unchecked
exception"
Yes, I meant using a runtime exception instead of a
checked
exception.
There was no consensus on putting the exception into
the
`TableResult`.
Regards,
Timo
On 01.04.20 15:35, Dawid Wysakowicz wrote:
When considering the multi-line support I think it
is
helpful
to
start
with a use case in mind. In my opinion consumers of
this method
will
be:
1. sql-client
2. third-part sql based platforms
@Godfrey As for the quit/source/... commands. I
think
those
belong
to
the responsibility of aforementioned. I think they
should not
be
understandable by the TableEnvironment. What would
quit
on a
TableEnvironment do? Moreover I think such commands
should be
prefixed
appropriately. I think it's a common practice to
e.g.
prefix
those
with
! or : to say they are meta commands of the tool
rather
than a
query.
I also don't necessarily understand why platform
users
need to
know
the
kind of the query to use the proposed method. They
should get
the
type
from the TableResult#ResultKind. If the ResultKind
is
SUCCESS,
it
was
a
DCL/DDL. If SUCCESS_WITH_CONTENT it was a DML/DQL.
If
that's
not
enough
we can enrich the TableResult with more explicit
kind
of query,
but
so
far I don't see such a need.
@Kurt In those cases I would assume the developers
want
to
present
results of the queries anyway. Moreover I think it
is
safe to
assume
they can adhere to such a contract that the results
must be
iterated.
For direct users of TableEnvironment/Table API this
method does
not
make
much sense anyway, in my opinion. I think we can
rather
safely
assume
in
this scenario they do not want to submit multiple
queries at a
single
time.
Best,
Dawid
On 01/04/2020 15:07, Kurt Young wrote:
One comment to `executeMultilineSql`, I'm afraid
sometimes
user
might
forget to
iterate the returned iterators, e.g. user submits a
bunch of
DDLs
and
expect the
framework will execute them one by one. But it
didn't.
Best,
Kurt
On Wed, Apr 1, 2020 at 5:10 PM Aljoscha Krettek<
aljos...@apache.org>
wrote:
Agreed to what Dawid and Timo said.
To answer your question about multi line SQL: no,
we
don't
think
we
need
this in Flink 1.11, we only wanted to make sure
that
the
interfaces
that
we now put in place will potentially allow this in
the
future.
Best,
Aljoscha
On 01.04.20 09:31, godfrey he wrote:
Hi, Timo & Dawid,
Thanks so much for the effort of `multiline
statements
supporting`,
I have a few questions about this method:
1. users can well control the execution logic
through the
proposed
method
if they know what the statements are (a
statement is a
DDL, a
DML
or
others).
but if a statement is from a file, that means
users
do not
know
what
the
statements are,
the execution behavior is unclear.
As a platform user, I think this method is hard
to
use,
unless
the
platform
defines
a set of rule about the statements order, such
as:
no select
in
the
middle,
dml must be at tail of sql file (which may be the
most case
in
product
env).
Otherwise the platform must parse the sql first,
then know
what
the
statements are.
If do like that, the platform can handle all
cases
through
`executeSql`
and
`StatementSet`.
2. SQL client can't also use
`executeMultilineSql`
to
supports
multiline
statements,
because there are some special commands
introduced in
SQL
client,
such as `quit`, `source`, `load jar` (not exist
now,
but
maybe
we
need
this
command
to support dynamic table source and udf).
Does TableEnvironment also supports those
commands?
3. btw, we must have this feature in
release-1.11?
I
find
there
are
few
user cases
in the feedback document which behavior is
unclear now.
regarding to "change the return value from
`Iterable<Row` to
`Iterator<Row`",
I couldn't agree more with this change. Just as
Dawid
mentioned
"The contract of the Iterable#iterator is that it
returns a
new
iterator
each time,
which effectively means we can iterate the
results
multiple
times.",
we does not provide iterate the results multiple
times.
If we want do that, the client must buffer all
results. but
it's
impossible
for streaming job.
Best,
Godfrey
Dawid Wysakowicz<dwysakow...@apache.org>
于2020年4月1日周三
上午3:14写道：
Thank you Timo for the great summary! It covers
(almost)
all
the
topics.
Even though in the end we are not suggesting
much
changes
to
the
current
state of FLIP I think it is important to lay out
all
possible
use
cases
so that we do not change the execution model
every
release.
There is one additional thing we discussed.
Could
we change
the
result
type of TableResult#collect to Iterator<Row>?
Even
though
those
interfaces do not differ much. I think Iterator
better
describes
that
the results might not be materialized on the
client
side,
but
can
be
retrieved on a per record basis. The contract of
the
Iterable#iterator
is that it returns a new iterator each time,
which
effectively
means
we
can iterate the results multiple times.
Iterating
the
results
is
not
possible when we don't retrieve all the results
from the
cluster
at
once.
I think we should also use Iterator for
TableEnvironment#executeMultilineSql(String
statements):
Iterator<TableResult>.
Best,
Dawid
On 31/03/2020 19:27, Timo Walther wrote:
Hi Godfrey,
Aljoscha, Dawid, Klou, and I had another
discussion around
FLIP-84.
In
particular, we discussed how the current status
of
the
FLIP
and
the
future requirements around multiline
statements,
async/sync,
collect()
fit together.
We also updated the FLIP-84 Feedback Summary
document [1]
with
some
use cases.
We believe that we found a good solution that
also
fits to
what
is
in
the current FLIP. So no bigger changes
necessary,
which is
great!
Our findings were:
1. Async vs sync submission of Flink jobs:
Having a blocking `execute()` in DataStream API
was
rather a
mistake.
Instead all submissions should be async because
this
allows
supporting
both modes if necessary. Thus, submitting all
queries
async
sounds
good to us. If users want to run a job sync,
they
can use
the
JobClient and wait for completion (or collect()
in
case of
batch
jobs).
2. Multi-statement execution:
For the multi-statement execution, we don't
see a
contradication
with
the async execution behavior. We imagine a
method
like:
TableEnvironment#executeMultilineSql(String
statements):
Iterable<TableResult>
Where the `Iterator#next()` method would
trigger
the next
statement
submission. This allows a caller to decide
synchronously
when
to
submit statements async to the cluster. Thus, a
service
such
as
the
SQL Client can handle the result of each
statement
individually
and
process statement by statement sequentially.
3. The role of TableResult and result retrieval
in
general
`TableResult` is similar to `JobClient`.
Instead
of
returning
a
`CompletableFuture` of something, it is a
concrete
util
class
where
some methods have the behavior of completable
future (e.g.
collect(),
print()) and some are already completed
(getTableSchema(),
getResultKind()).
`StatementSet#execute()` returns a single
`TableResult`
because
the
order is undefined in a set and all statements
have the
same
schema.
Its `collect()` will return a row for each
executed
`INSERT
INTO` in
the order of statement definition.
For simple `SELECT * FROM ...`, the query
execution might
block
until
`collect()` is called to pull buffered rows
from
the job
(from
socket/REST API what ever we will use in the
future). We
can
say
that
a statement finished successfully, when the
`collect#Iterator#hasNext`
has returned false.
I hope this summarizes our discussion
@Dawid/Aljoscha/Klou?
It would be great if we can add these findings
to
the FLIP
before we
start voting.
One minor thing: some `execute()` methods still
throw a
checked
exception; can we remove that from the FLIP?
Also
the
above
mentioned
`Iterator#next()` would trigger an execution
without
throwing
a
checked exception.
Thanks,
Timo
[1]
https://docs.google.com/document/d/1ueLjQWRPdLTFB_TReAyhseAX-1N3j4WYWD0F02Uau0E/edit#
On 31.03.20 06:28, godfrey he wrote:
Hi, Timo & Jark
Thanks for your explanation.
Agree with you that async execution should
always
be
async,
and sync execution scenario can be covered by
async
execution.
It helps provide an unified entry point for
batch
and
streaming.
I think we can also use sync execution for
some
testing.
So, I agree with you that we provide
`executeSql`
method
and
it's
async
method.
If we want sync method in the future, we can
add
method
named
`executeSqlSync`.
I think we've reached an agreement. I will
update
the
document,
and
start
voting process.
Best,
Godfrey
Jark Wu<imj...@gmail.com> 于2020年3月31日周二
上午12:46写道：
Hi,
I didn't follow the full discussion.
But I share the same concern with Timo that
streaming
queries
should
always
be async.
Otherwise, I can image it will cause a lot of
confusion
and
problems
if
users don't deeply keep the "sync" in mind
(e.g.
client
hangs).
Besides, the streaming mode is still the
majority use
cases
of
Flink
and
Flink SQL. We should put the usability at a
high
priority.
Best,
Jark
On Mon, 30 Mar 2020 at 23:27, Timo Walther<
twal...@apache.org>
wrote:
Hi Godfrey,
maybe I wasn't expressing my biggest concern
enough in
my
last
mail.
Even in a singleline and sync execution, I
think that
streaming
queries
should not block the execution. Otherwise it
is
not
possible
to
call
collect() or print() on them afterwards.
"there are too many things need to discuss
for
multiline":
True, I don't want to solve all of them
right
now. But
what
I
know
is
that our newly introduced methods should fit
into a
multiline
execution.
There is no big difference of calling
`executeSql(A),
executeSql(B)` and
processing a multiline file `A;\nB;`.
I think the example that you mentioned can
simply be
undefined
for
now.
Currently, no catalog is modifying data but
just
metadata.
This
is a
separate discussion.
"result of the second statement is
indeterministic":
Sure this is indeterministic. But this is
the
implementers
fault
and we
cannot forbid such pipelines.
How about we always execute streaming
queries
async? It
would
unblock
executeSql() and multiline statements.
Having a `executeSqlAsync()` is useful for
batch.
However,
I
don't
want
`sync/async` be the new batch/stream flag.
The
execution
behavior
should
come from the query itself.
Regards,
Timo
On 30.03.20 11:12, godfrey he wrote:
Hi Timo,
Agree with you that streaming queries is
our
top
priority,
but I think there are too many things need
to
discuss
for
multiline
statements:
e.g.
1. what's the behaivor of DDL and DML
mixing
for async
execution:
create table t1 xxx;
create table t2 xxx;
insert into t2 select * from t1 where xxx;
drop table t1; // t1 may be a MySQL table,
the
data
will
also be
deleted.
t1 is dropped when "insert" job is running.
2. what's the behaivor of unified scenario
for
async
execution:
(as you
mentioned)
INSERT INTO t1 SELECT * FROM s;
INSERT INTO t2 SELECT * FROM s JOIN t1 EMIT
STREAM;
The result of the second statement is
indeterministic,
because
the
first
statement maybe is running.
I think we need to put a lot of effort to
define the
behavior of
logically
related queries.
In this FLIP, I suggest we only handle
single
statement,
and
we
also
introduce an async execute method
which is more important and more often used
for users.
Dor the sync methods (like
`TableEnvironment.executeSql`
and
`StatementSet.execute`),
the result will be returned until the job
is
finished.
The
following
methods will be introduced in this FLIP:
/**
* Asynchronously execute the given
single
statement
*/
TableEnvironment.executeSqlAsync(String
statement):
TableResult
/**
* Asynchronously execute the dml
statements as
a
batch
*/
StatementSet.executeAsync(): TableResult
public interface TableResult {
/**
* return JobClient for DQL and
DML
in async
mode,
else
return
Optional.empty
*/
Optional<JobClient>
getJobClient();
}
what do you think?
Best,
Godfrey
Timo Walther<twal...@apache.org>
于2020年3月26日周四
下午9:15写道：
Hi Godfrey,
executing streaming queries must be our
top
priority
because
this
is
what distinguishes Flink from competitors.
If
we
change
the
execution
behavior, we should think about the other
cases as
well
to
not
break
the
API a third time.
I fear that just having an async execute
method will
not
be
enough
because users should be able to mix
streaming
and
batch
queries
in a
unified scenario.
If I remember it correctly, we had some
discussions
in
the
past
about
what decides about the execution mode of a
query.
Currently, we
would
like to let the query decide, not derive
it
from the
sources.
So I could image a multiline pipeline as:
USE CATALOG 'mycat';
INSERT INTO t1 SELECT * FROM s;
INSERT INTO t2 SELECT * FROM s JOIN t1
EMIT
STREAM;
For executeMultilineSql():
sync because regular SQL
sync because regular Batch SQL
async because Streaming SQL
For executeAsyncMultilineSql():
async because everything should be async
async because everything should be async
async because everything should be async
What we should not start for
executeAsyncMultilineSql():
sync because DDL
async because everything should be async
async because everything should be async
What are you thoughts here?
Regards,
Timo
On 26.03.20 12:50, godfrey he wrote:
Hi Timo,
I agree with you that streaming queries
mostly need
async
execution.
In fact, our original plan is only
introducing sync
methods in
this
FLIP,
and async methods (like
"executeSqlAsync")
will be
introduced
in
the
future
which is mentioned in the appendix.
Maybe the async methods also need to be
considered
in
this
FLIP.
I think sync methods is also useful for
streaming
which
