This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new da3b56b16c2 [FLINK-41032][python] Preserve column dtypes when
converting empty tables to pandas (#29526)
da3b56b16c2 is described below
commit da3b56b16c22d7d2920deb919a4d1c9cb72cc4ae
Author: Biao Geng <[email protected]>
AuthorDate: Tue Oct 27 15:44:57 2026 +0800
[FLINK-41032][python] Preserve column dtypes when converting empty tables
to pandas (#29526)
---
flink-python/pyflink/table/table.py | 42 ++++++++-------
.../pyflink/table/tests/test_pandas_conversion.py | 59 +++++++++++++++++++++-
2 files changed, 83 insertions(+), 18 deletions(-)
diff --git a/flink-python/pyflink/table/table.py
b/flink-python/pyflink/table/table.py
index 6195d560817..8b521df8a0a 100644
--- a/flink-python/pyflink/table/table.py
+++ b/flink-python/pyflink/table/table.py
@@ -16,8 +16,10 @@
# limitations under the License.
################################################################################
+from datetime import tzinfo
+
from py4j.java_gateway import get_method
-from typing import Union, TYPE_CHECKING
+from typing import Optional, Union, TYPE_CHECKING
if TYPE_CHECKING:
import pandas
@@ -32,7 +34,7 @@ from pyflink.table.table_pipeline import TablePipeline
from pyflink.table.table_result import TableResult
from pyflink.table.table_schema import TableSchema
from pyflink.table.catalog import ResolvedSchema
-from pyflink.table.types import create_arrow_schema
+from pyflink.table.types import LocalZonedTimestampType, create_arrow_schema
from pyflink.table.udf import UserDefinedScalarFunctionWrapper, \
UserDefinedAggregateFunctionWrapper, UserDefinedTableFunctionWrapper, \
UserDefinedAsyncScalarFunctionWrapper
@@ -950,25 +952,31 @@ class Table(object):
batches_iterator =
gateway.jvm.org.apache.flink.table.runtime.arrow.ArrowUtils\
.collectAsPandasDataFrame(self._j_table, max_arrow_batch_size)
schema = self.get_schema()
- if batches_iterator.hasNext():
+ field_types = schema.get_field_data_types()
+ has_rows = batches_iterator.hasNext()
+ arrow_schema = create_arrow_schema(
+ schema.get_field_names(), field_types, allow_nested=not has_rows)
+ timezone: Optional[tzinfo] = None
+ if any(isinstance(field_type, LocalZonedTimestampType) for field_type
in field_types):
import pytz
- timezone = pytz.timezone(
-
self._j_table.getTableEnvironment().getConfig().getLocalTimeZone().getId())
- serializer = ArrowSerializer(
- create_arrow_schema(schema.get_field_names(),
schema.get_field_data_types()),
- schema.to_row_data_type(),
- timezone)
+ j_timezone =
self._j_table.getTableEnvironment().getConfig().getLocalTimeZone()
+ timezone_id = j_timezone.getId()
+ if timezone_id.startswith(('GMT+', 'GMT-')):
+ timezone =
pytz.FixedOffset(j_timezone.normalized().getTotalSeconds() // 60)
+ else:
+ timezone = pytz.timezone(timezone_id)
+ if has_rows:
+ serializer = ArrowSerializer(arrow_schema,
schema.to_row_data_type(), timezone)
import pyarrow as pa
table =
pa.Table.from_batches(serializer.load_from_iterator(batches_iterator))
- pdf = table.to_pandas()
-
- for field_name in schema.get_field_names():
- pdf[field_name] = tz_convert_from_internal(
- pdf[field_name], schema.get_field_data_type(field_name),
timezone)
- return pdf
else:
- import pandas as pd
- return pd.DataFrame.from_records([],
columns=schema.get_field_names())
+ table = arrow_schema.empty_table()
+ pdf = table.to_pandas()
+
+ for field_name in schema.get_field_names():
+ pdf[field_name] = tz_convert_from_internal(
+ pdf[field_name], schema.get_field_data_type(field_name),
timezone)
+ return pdf
@Deprecated(since="2.1.0", detail="Use :func:`Table.get_resolved_schema`
instead.")
def get_schema(self) -> TableSchema:
diff --git a/flink-python/pyflink/table/tests/test_pandas_conversion.py
b/flink-python/pyflink/table/tests/test_pandas_conversion.py
index 8622f2ccc6a..c359e530aeb 100644
--- a/flink-python/pyflink/table/tests/test_pandas_conversion.py
+++ b/flink-python/pyflink/table/tests/test_pandas_conversion.py
@@ -185,7 +185,64 @@ class PandasConversionITTests(PandasConversionTestBase):
def test_empty_to_pandas(self):
table = self.t_env.from_pandas(self.pdf, self.data_type)
pdf = table.filter(table.f1 < 0).to_pandas()
- self.assertTrue(pdf.empty)
+ assert_frame_equal(pdf, table.to_pandas().iloc[:0])
+
+ def test_to_pandas_with_nested_types(self):
+ import pandas as pd
+
+ for expression, error in [('ARRAY[ROW(1)]', ValueError),
('ROW(ROW(1))', TypeError)]:
+ with self.subTest(expression=expression):
+ query = 'SELECT 1 AS a, %s AS payload FROM (VALUES (1)) AS
t(x)' % expression
+ result = self.t_env.sql_query(query + ' WHERE
FALSE').to_pandas()
+ expected = pd.DataFrame({'a': pd.Series(dtype='int32'),
+ 'payload': pd.Series(dtype='object')})
+ assert_frame_equal(result, expected)
+ with self.assertRaisesRegex(error, 'ROW is not supported'):
+ self.t_env.sql_query(query).to_pandas()
+
+ def test_empty_to_pandas_with_timestamps(self):
+ original_timezone = self.t_env.get_config().get_local_timezone()
+ self.addCleanup(self.t_env.get_config().set_local_timezone,
original_timezone)
+ for timezone in ['UTC', 'Asia/Shanghai', 'Etc/GMT-8']:
+ with self.subTest(timezone=timezone):
+ self.t_env.get_config().set_local_timezone(timezone)
+ table = self.t_env.sql_query("""
+ SELECT
+ 1 AS a,
+ CAST(TIMESTAMP '2024-01-01 00:00:00.123456789'
+ AS TIMESTAMP(0)) AS ts_seconds,
+ CAST(TIMESTAMP '2024-01-01 00:00:00.123456789'
+ AS TIMESTAMP(3)) AS ts_millis,
+ CAST(TIMESTAMP '2024-01-01 00:00:00.123456789'
+ AS TIMESTAMP(6)) AS ts_micros,
+ TIMESTAMP '2024-01-01 00:00:00.123456789' AS ts_nanos,
+ TO_TIMESTAMP_LTZ(1704067200123, 3) AS ltz
+ """)
+ expected = table.to_pandas().iloc[:0]
+ result = table.filter(table.a < 0).to_pandas()
+ assert_frame_equal(result, expected)
+ self.assertEqual(timezone, str(result['ltz'].dt.tz))
+
+ def test_to_pandas_with_fixed_offset_timezones(self):
+ original_timezone = self.t_env.get_config().get_local_timezone()
+ self.addCleanup(self.t_env.get_config().set_local_timezone,
original_timezone)
+ for timezone, named_timezone, offset_seconds in [
+ ('GMT+08:00', 'Asia/Shanghai', 28800),
+ ('GMT+05:30', 'Asia/Kolkata', 19800),
+ ('GMT-03:30', 'America/St_Johns', -12600)]:
+ with self.subTest(timezone=timezone):
+ self.t_env.get_config().set_local_timezone(named_timezone)
+ table = self.t_env.sql_query("""
+ SELECT 1 AS a, TO_TIMESTAMP_LTZ(1704067200123, 3) AS ltz
+ """)
+ expected_timestamp = table.to_pandas()['ltz'].iloc[0]
+ self.t_env.get_config().set_local_timezone(timezone)
+ result = table.to_pandas()
+ timestamp = result['ltz'].iloc[0]
+ self.assertEqual(expected_timestamp, timestamp)
+ self.assertEqual(datetime.timedelta(seconds=offset_seconds),
timestamp.utcoffset())
+ empty_result = table.filter(table.a < 0).to_pandas()
+ assert_frame_equal(empty_result, result.iloc[:0])
def test_to_pandas_preserves_local_zoned_timestamp_instants(self):
import pandas as pd
