anew commented on code in PR #58843:
URL: https://github.com/apache/spark/pull/58843#discussion_r4044701205
##########
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:
Added the ignore_null_updates=True + non-empty except-list case, and packed
the conflict/empty-list checks into two subTest grid tests
--
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]
