You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/08/03 06:13:02 UTC
[GitHub] [flink] dianfu commented on a change in pull request #12902: [FLINK-18415][python] Support TableResult#collect in the Python Table…
dianfu commented on a change in pull request #12902:
URL: https://github.com/apache/flink/pull/12902#discussion_r464204272
##########
File path: flink-python/pyflink/table/table_result.py
##########
@@ -130,6 +132,18 @@ def get_result_kind(self):
"""
return ResultKind._from_j_result_kind(self._j_table_result.getResultKind())
+ def collect(self):
+ """
+ Get the result contents as a closeable row iterator.
+
+ :return: A CloseableIterator.
+ """
+ data_types = self._j_table_result.getTableSchema().getFieldDataTypes()
+
+ j_iter = self._j_table_result.collect()
+
Review comment:
Add the version information:
```
.. versionadded:: 1.12.0
```
##########
File path: flink-python/pyflink/table/tests/test_table_api.py
##########
@@ -31,6 +32,20 @@ def test_execute(self):
source.execute()
# TODO check result
+ def test_collect(self):
+ self.t_env.get_config().get_configuration().set_string(
+ "taskmanager.memory.task.off-heap.size", "80mb")
+ element_data = [({"a": [1, 2]}, [1, 2], Row("a")(1)), ({"a": [3, 4]}, [3, 4], Row("a")(1))]
Review comment:
could you improve the tests to cover all the data types supported?
##########
File path: flink-python/pyflink/table/utils.py
##########
@@ -67,3 +71,60 @@ def tz_convert_to_internal(s, t: DataType, local_tz):
elif is_datetime64tz_dtype(s.dtype):
return s.dt.tz_convert(local_tz).dt.tz_localize(None)
return s
+
+
+class CloseableIterator(object):
+ """
+ Representing an Iterator that is also auto closeable.
+ """
+ def __init__(self, j_closeable_iterator, result_types):
+ self._j_closeable_iterator = j_closeable_iterator
+ self._j_result_types = result_types
Review comment:
result_types -> field_data_types
##########
File path: flink-python/pyflink/table/tests/test_table_api.py
##########
@@ -31,6 +32,20 @@ def test_execute(self):
source.execute()
# TODO check result
+ def test_collect(self):
+ self.t_env.get_config().get_configuration().set_string(
+ "taskmanager.memory.task.off-heap.size", "80mb")
+ element_data = [({"a": [1, 2]}, [1, 2], Row("a")(1)), ({"a": [3, 4]}, [3, 4], Row("a")(1))]
+ source = self.t_env.from_elements(element_data, ["a", "b", "c"])
+ result_table = self.t_env.execute_sql("SELECT * FROM %s" % source)
Review comment:
What happens for a Table with retraction rows? Could you add a test for it?
##########
File path: flink-python/pyflink/table/tests/test_table_api.py
##########
@@ -31,6 +32,20 @@ def test_execute(self):
source.execute()
# TODO check result
+ def test_collect(self):
+ self.t_env.get_config().get_configuration().set_string(
+ "taskmanager.memory.task.off-heap.size", "80mb")
+ element_data = [({"a": [1, 2]}, [1, 2], Row("a")(1)), ({"a": [3, 4]}, [3, 4], Row("a")(1))]
+ source = self.t_env.from_elements(element_data, ["a", "b", "c"])
+ result_table = self.t_env.execute_sql("SELECT * FROM %s" % source)
Review comment:
this line could be removed as source is already a Table
##########
File path: flink-python/pyflink/table/types.py
##########
@@ -2405,6 +2405,10 @@ def to_arrow_type(data_type):
raise ValueError("field_type %s is not supported." % data_type)
+def get_data_type_from_java_type(j_java_types):
Review comment:
What about remove this method?
##########
File path: flink-python/src/main/java/org/apache/flink/api/common/python/PythonBridgeUtils.java
##########
@@ -175,4 +221,20 @@ private static void initialize() {
}
}
}
+
+ private static int getOffsetFromLocalMillis(long millisLocal) {
Review comment:
where it's used?
##########
File path: flink-python/pyflink/table/table_result.py
##########
@@ -130,6 +132,18 @@ def get_result_kind(self):
"""
return ResultKind._from_j_result_kind(self._j_table_result.getResultKind())
+ def collect(self):
+ """
+ Get the result contents as a closeable row iterator.
+
+ :return: A CloseableIterator.
+ """
+ data_types = self._j_table_result.getTableSchema().getFieldDataTypes()
Review comment:
data_types -> field_data_types?
##########
File path: flink-python/pyflink/table/tests/test_table_api.py
##########
@@ -31,6 +32,20 @@ def test_execute(self):
source.execute()
# TODO check result
+ def test_collect(self):
+ self.t_env.get_config().get_configuration().set_string(
Review comment:
Could be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org