Skip to content

[FLINK-38797][python] Fixed PyFlink issue CsvSchemaBuilder.set_null_value missing return self #27331

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wchan87 wants to merge 2 commits into from
Closed

[FLINK-38797][python] Fixed PyFlink issue CsvSchemaBuilder.set_null_value missing return self #27331

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d1bc39b
FLINK-38797 Fixed PyFlink issue CsvSchemaBuilder.set_null_value missi…
wchan87 Dec 10, 2025
f7fc5bf
[FLINK-38797][python] Updated test_csv.py method name to pass flake8 …
wchan87 Dec 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions flink-python/pyflink/datastream/formats/csv.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,7 @@ def set_null_value(self, null_value: str):
Set literal for null value, default to empty sequence.
"""
self._j_schema_builder.setNullValue(null_value)
return self

def disable_quote_char(self):
"""
Expand Down
24 changes: 24 additions & 0 deletions flink-python/pyflink/datastream/formats/tests/test_csv.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,12 @@ def test_csv_strict_headers(self):
self.env.execute('test_csv_strict_headers')
_check_csv_strict_headers_results(self, self.test_sink.get_results(True, False))

def test_csv_default_null_value(self):
schema, lines = _create_csv_default_null_value_schema_and_lines()
self._build_csv_job(schema, lines)
self.env.execute('test_csv_default_null_value')
_check_csv_default_null_value_results(self, self.test_sink.get_results(True, False))

def test_csv_default_quote_char(self):
schema, lines = _create_csv_default_quote_char_schema_and_lines()
self._build_csv_job(schema, lines)
Expand Down Expand Up @@ -344,6 +350,24 @@ def _check_csv_use_header_results(test, results):
test.assertEqual(row['number'], 123)


def _create_csv_default_null_value_schema_and_lines() -> Tuple[CsvSchema, List[str]]:
schema = CsvSchema.builder() \
.add_string_column('string') \
.add_number_column('number') \
.set_null_value('') \
Copy link
Contributor

@davidradl davidradl Dec 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good. Some testing suggestions:
I would test a test with .set_null_value(), set the default? If so we should test that
I also suggest a test specifying that a non default literal for null works.

Copy link
Contributor

@dianfu dianfu Dec 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A null_value must be specified if .set_null_value is called. There is no default value.

.build()
lines = [
',123\n'
]
return schema, lines


def _check_csv_default_null_value_results(test, results):
row = results[0]
test.assertEqual(row['string'], None)
test.assertEqual(row['number'], 123)


def _create_csv_strict_headers_schema_and_lines() -> Tuple[CsvSchema, List[str]]:
schema = CsvSchema.builder() \
.add_string_column('string') \
Expand Down