AnishMahto commented on code in PR #58843:
URL: https://github.com/apache/spark/pull/58843#discussion_r4028893934
##########
python/pyspark/pipelines/api.py:
##########
@@ -587,6 +590,19 @@ def create_auto_cdc_flow(
this list will be in the output table.
:param stored_as_scd_type: The SCD type for the target table. 1 (or "1")
and 2 (or "2") are \
supported. When not specified, the server default applies.
+ :param ignore_null_updates: When True, null values in an incoming update
are ignored and the \
+ existing target value is preserved, for every column. Defaults to
False, in which case \
+ nulls overwrite the target. Mutually exclusive with
ignore_null_updates_column_list and \
+ ignore_null_updates_except_column_list.
+ :param ignore_null_updates_column_list: The subset of columns for which
null values in an \
+ incoming update are ignored. This should be a list of column
identifiers without \
+ qualifiers, expressed as either Python strings or PySpark Columns.
Mutually exclusive \
+ with ignore_null_updates and ignore_null_updates_except_column_list.
+ :param ignore_null_updates_except_column_list: The subset of columns for
which null values \
+ in an incoming update overwrite the target; nulls are ignored for all
other columns. \
+ This should be a list of column identifiers without qualifiers,
expressed as either \
+ Python strings or PySpark Columns. Mutually exclusive with
ignore_null_updates and \
+ ignore_null_updates_column_list.
Review Comment:
Should we explicitly document that a if `ignore_null_updates_column_list` or
`ignore_null_updates_except_column_list` is specified, then
`ignore_null_updates = False` is considered a no-op/unspecified?
##########
python/pyspark/pipelines/tests/test_auto_cdc_flow.py:
##########
@@ -298,6 +298,200 @@ def
test_create_auto_cdc_flow_rejects_both_track_history_column_list_and_except(
)
self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+ def test_create_auto_cdc_flow_with_ignore_null_updates(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertTrue(flow.ignore_null_updates)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def test_create_auto_cdc_flow_ignore_null_updates_defaults(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertFalse(flow.ignore_null_updates)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def test_create_auto_cdc_flow_with_ignore_null_updates_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=["val"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertFalse(flow.ignore_null_updates)
+ assert flow.ignore_null_updates_column_list is not None
+ self.assertEqual(len(flow.ignore_null_updates_column_list), 1)
+ self.assertIsInstance(flow.ignore_null_updates_column_list[0], Column)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def
test_create_auto_cdc_flow_with_ignore_null_updates_except_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_except_column_list=["op", "seq"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ assert flow.ignore_null_updates_except_column_list is not None
+ self.assertEqual(len(flow.ignore_null_updates_except_column_list), 2)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_except_with_column_list(self):
+ # An empty except list is a specified selection ("ignore nulls on all
columns"), so it
+ # may not coexist with an include list.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=["val"],
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_empty_except_list(self):
+ # An empty (specified) except list may not coexist with the
ignore_null_updates flag.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_empty_column_list(self):
+ # An empty (specified) include list may not coexist with the
ignore_null_updates flag.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ ignore_null_updates_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_both_ignore_null_updates_column_list_and_except(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ ignore_null_updates_column_list=["val"],
+ ignore_null_updates_except_column_list=["op"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ ignore_null_updates=True,
+ ignore_null_updates_column_list=["val"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def test_create_auto_cdc_flow_rejects_non_bool_ignore_null_updates(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkTypeError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates="yes", # type: ignore[arg-type]
+ )
+ self.assertEqual(ctx.exception.getCondition(), "NOT_EXPECTED_TYPE")
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_column_list(self):
+ # A lone empty include list ignores nulls for no columns and cannot be
represented on the
+ # wire, so it is rejected.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(), "CANNOT_BE_EMPTY")
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_except_column_list(self):
+ # An empty except list cannot be represented on the wire, so it is
rejected (symmetric
+ # with the include list); use ignore_null_updates=True to ignore nulls
on all columns.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(), "CANNOT_BE_EMPTY")
Review Comment:
Consider using `PySparkErrorTestUtils.check_error`
##########
python/pyspark/pipelines/tests/test_auto_cdc_flow.py:
##########
@@ -298,6 +298,200 @@ def
test_create_auto_cdc_flow_rejects_both_track_history_column_list_and_except(
)
self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+ def test_create_auto_cdc_flow_with_ignore_null_updates(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertTrue(flow.ignore_null_updates)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def test_create_auto_cdc_flow_ignore_null_updates_defaults(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertFalse(flow.ignore_null_updates)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def test_create_auto_cdc_flow_with_ignore_null_updates_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=["val"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ self.assertFalse(flow.ignore_null_updates)
+ assert flow.ignore_null_updates_column_list is not None
+ self.assertEqual(len(flow.ignore_null_updates_column_list), 1)
+ self.assertIsInstance(flow.ignore_null_updates_column_list[0], Column)
+ self.assertIsNone(flow.ignore_null_updates_except_column_list)
+
+ def
test_create_auto_cdc_flow_with_ignore_null_updates_except_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_except_column_list=["op", "seq"],
+ )
+
+ flow = cast(AutoCdcFlow, registry.auto_cdc_flows[0])
+ assert flow.ignore_null_updates_except_column_list is not None
+ self.assertEqual(len(flow.ignore_null_updates_except_column_list), 2)
+ self.assertIsNone(flow.ignore_null_updates_column_list)
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_except_with_column_list(self):
+ # An empty except list is a specified selection ("ignore nulls on all
columns"), so it
+ # may not coexist with an include list.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=["val"],
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_empty_except_list(self):
+ # An empty (specified) except list may not coexist with the
ignore_null_updates flag.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_empty_column_list(self):
+ # An empty (specified) include list may not coexist with the
ignore_null_updates flag.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates=True,
+ ignore_null_updates_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_both_ignore_null_updates_column_list_and_except(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ ignore_null_updates_column_list=["val"],
+ ignore_null_updates_except_column_list=["op"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def
test_create_auto_cdc_flow_rejects_ignore_null_updates_with_column_list(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("tgt")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="tgt",
+ source="src",
+ keys=[col("id")],
+ sequence_by=expr("ts"),
+ ignore_null_updates=True,
+ ignore_null_updates_column_list=["val"],
+ )
+ self.assertEqual(ctx.exception.getCondition(),
"CANNOT_SET_TOGETHER")
+
+ def test_create_auto_cdc_flow_rejects_non_bool_ignore_null_updates(self):
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkTypeError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates="yes", # type: ignore[arg-type]
+ )
+ self.assertEqual(ctx.exception.getCondition(), "NOT_EXPECTED_TYPE")
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_column_list(self):
+ # A lone empty include list ignores nulls for no columns and cannot be
represented on the
+ # wire, so it is rejected.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(), "CANNOT_BE_EMPTY")
+
+ def
test_create_auto_cdc_flow_rejects_empty_ignore_null_updates_except_column_list(self):
+ # An empty except list cannot be represented on the wire, so it is
rejected (symmetric
+ # with the include list); use ignore_null_updates=True to ignore nulls
on all columns.
+ registry = LocalGraphElementRegistry()
+ with graph_element_registration_context(registry):
+ dp.create_streaming_table("t")
+ with self.assertRaises(PySparkValueError) as ctx:
+ dp.create_auto_cdc_flow(
+ target="t",
+ source="s",
+ keys=[col("k")],
+ sequence_by=expr("seq"),
+ ignore_null_updates_except_column_list=[],
+ )
+ self.assertEqual(ctx.exception.getCondition(), "CANNOT_BE_EMPTY")
+
Review Comment:
We should also assert that specifying `ignore_null_updates = True` but
non-empty ignore-null/exclude lists is invalid.
Also can we pack several of the same-validation tests into one grid test
(ex. using `subTest`)?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
