[FLINK-19417][python] Fix the bug of the method from_data_stream in table_environement #13491
Conversation
|
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community Automated ChecksLast check on commit f998eec (Fri Feb 19 07:28:29 UTC 2021) Warnings:
Mention the bot in a comment to re-run the automated checks. Review Progress
Please see the Pull Request Review Guide for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commandsThe @flinkbot bot supports the following commands:
|
…able_environement
There was a problem hiding this comment.
@SteNicholas Thanks a lot for the fix. Good Work. I only left a minor comments.
| j_table = self._j_tenv.fromDataStream(data_stream._j_data_stream) | ||
| return Table(j_table=j_table, t_env=self._j_tenv) | ||
| elif len(fields) == 1 and isinstance(fields[0], str): | ||
| j_table = self._j_tenv.fromDataStream(data_stream._j_data_stream, fields[0]) |
There was a problem hiding this comment.
What about adding a warn to tell the user that this method has been deprecated
There was a problem hiding this comment.
@HuangXingBo Yes, I would like to add warn log to tell users with this deprecated method.
| to_jarray(gateway.jvm.Expression, | ||
| [_get_java_expression(f) | ||
| for f in fields])) | ||
| return None if j_table is None else Table(j_table=j_table, t_env=self) |
There was a problem hiding this comment.
What about raising an Exception to tell the user that the parameter is wrong, instead of returning a None Table
There was a problem hiding this comment.
I agree with @HuangXingBo, could refer to Table.select as an example.
There was a problem hiding this comment.
@dianfu @HuangXingBo OK, I will modify this return value as you mentioned.
| elif len(fields) == 1 and isinstance(fields[0], str): | ||
| j_table = self._j_tenv.fromDataStream(data_stream._j_data_stream, fields[0]) | ||
| elif len(fields) > 0 and \ | ||
| [isinstance(f, Expression) for f in fields] == [True] * len(fields): |
There was a problem hiding this comment.
what about all(isinstance(f, Expression) for f in fields)
| [isinstance(f, Expression) for f in fields] == [True] * len(fields): | ||
| gateway = get_gateway() | ||
| j_table = self._j_tenv.fromDataStream(data_stream._j_data_stream, | ||
| to_jarray(gateway.jvm.Expression, |
There was a problem hiding this comment.
can use to_expression_jarray
| to_jarray(gateway.jvm.Expression, | ||
| [_get_java_expression(f) | ||
| for f in fields])) | ||
| return None if j_table is None else Table(j_table=j_table, t_env=self) |
There was a problem hiding this comment.
I agree with @HuangXingBo, could refer to Table.select as an example.
…able_environement
|
@HuangXingBo @dianfu I have already followed up with your comments. Please review this again. |
|
@SteNicholas Thanks for the update. LGTM. There are check style issues. Could you take a look at? |
…able_environement
|
@SteNicholas Thanks a lot for the update. LGTM. |
What is the purpose of the change
The parameter of method
from_data_streaminStreamTableEnvironmentfieldsshould be str or expression, not the current list [str]. And thetable_envobject passed to the Table object should be Python's TableEnvironment, not Java's TableEnvironment.Brief change log
from_data_streaminStreamTableEnvironmentfieldsto str or expression .Verifying this change
test_from_data_streaminTableEnvironmentTestfor the case that verifies the creation of data stream byfrom_data_streamwith expression.Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes / no)Documentation