Skip to content

Commit b9206d3

Browse files
zhuanshenbsj1zhuanshenbsj1
committed
mirror fix
1 parent c0ff23d commit b9206d3

2 files changed

Lines changed: 230 additions & 2 deletions

File tree

‎paimon-python/pypaimon/schema/schema_manager.py‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -442,6 +442,8 @@ def _validate_blob_fields(
442442
f"'{CoreOptions.VIDEO_FRAME_FIELD.key()}' only supports append-only tables."
443443
)
444444

445+
_validate_sequence_group_ordering_fields(fields, options, core_options, primary_keys)
446+
445447
if blob_field_names:
446448
if blob_field_names.intersection(partition_keys):
447449
raise ValueError(
@@ -479,6 +481,54 @@ def _validate_blob_fields(
479481
)
480482

481483

484+
def _validate_sequence_group_ordering_fields(
485+
fields: List[DataField],
486+
options: dict,
487+
core_options: CoreOptions,
488+
primary_keys: List[str],
489+
):
490+
"""Mirror Java SchemaValidation#validateSequenceGroupOrderingFields and the
491+
primary-key checks in PartialUpdateMergeFunction.Factory."""
492+
from pypaimon.common.options.core_options import MergeEngine
493+
494+
if core_options.merge_engine() != MergeEngine.PARTIAL_UPDATE:
495+
return
496+
497+
field_map = {field.name: field for field in fields}
498+
for option_key, option_val in options.items():
499+
if not (
500+
option_key.startswith("fields.")
501+
and option_key.endswith(".sequence-group")
502+
and len(option_key) > len("fields..sequence-group")
503+
):
504+
continue
505+
506+
ordering_fields_str = option_key[len("fields."): -len(".sequence-group")]
507+
for field_name in ordering_fields_str.split(","):
508+
field_name = field_name.strip()
509+
field = field_map.get(field_name)
510+
if field is not None and _contains_blob_type(field.type):
511+
raise ValueError(
512+
f"Field '{field_name}' with type {field.type} cannot be used as a "
513+
f"sequence-group ordering field in option '{option_key}'."
514+
)
515+
if field_name in primary_keys:
516+
raise ValueError(
517+
f"The sequence-group '{option_key}' contains primary key field "
518+
f"'{field_name}', which is not allowed. Primary key columns "
519+
f"cannot be put in sequence-group."
520+
)
521+
522+
for protected_field in str(option_val or "").split(","):
523+
protected_field = protected_field.strip()
524+
if protected_field and protected_field in primary_keys:
525+
raise ValueError(
526+
f"The sequence-group '{option_key}' contains primary key field "
527+
f"'{protected_field}', which is not allowed. Primary key columns "
528+
f"cannot be put in sequence-group."
529+
)
530+
531+
482532
def _validate_primary_key_blob_fields(
483533
fields: List[DataField],
484534
options: dict,
@@ -491,6 +541,7 @@ def _validate_primary_key_blob_fields(
491541
"""Mirror Java SchemaValidation#validatePrimaryKeyBlobKeyConfiguration and
492542
validatePrimaryKeyBlobConfiguration."""
493543
from pypaimon.common.options.core_options import ChangelogProducer, MergeEngine
544+
from pypaimon.common.options.options_utils import OptionsUtils
494545

495546
managed_blob_fields = [
496547
field.name
@@ -546,6 +597,46 @@ def _validate_primary_key_blob_fields(
546597
"Primary-key managed BLOB tables do not support 'pk-clustering-override'."
547598
)
548599

600+
if (
601+
core_options.merge_engine() == MergeEngine.PARTIAL_UPDATE
602+
and not core_options.ignore_delete()
603+
):
604+
fields_protected_by_sequence_group = set()
605+
for key, val in options.items():
606+
if (
607+
key.startswith("fields.")
608+
and key.endswith(".sequence-group")
609+
and len(key) > len("fields..sequence-group")
610+
):
611+
for protected_field in str(val or "").split(","):
612+
protected_field = protected_field.strip()
613+
if protected_field:
614+
fields_protected_by_sequence_group.add(protected_field)
615+
616+
for field in managed_blob_fields:
617+
if field not in fields_protected_by_sequence_group:
618+
continue
619+
agg_func = options.get(f"fields.{field}.aggregate-function")
620+
if agg_func is None:
621+
agg_func = options.get("fields.default-aggregate-function")
622+
623+
ignore_retract_raw = options.get(f"fields.{field}.ignore-retract")
624+
ignore_retract = (
625+
OptionsUtils.convert_to_boolean(ignore_retract_raw)
626+
if ignore_retract_raw is not None
627+
else False
628+
)
629+
if (
630+
agg_func is not None
631+
and agg_func != "last_value"
632+
and not ignore_retract
633+
):
634+
raise ValueError(
635+
f"Managed BLOB field '{field}' cannot use aggregate function '{agg_func}' "
636+
f"because managed BLOB payloads are not retained in retract messages. "
637+
f"Set 'fields.{field}.ignore-retract' to true to ignore retract messages."
638+
)
639+
549640
if blob_field_names.intersection(primary_keys):
550641
raise ValueError(
551642
"The BLOB, ARRAY<BLOB> or MAP<X, BLOB> type column can not be "

‎paimon-python/pypaimon/tests/schema_manager_test.py‎

Lines changed: 139 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -235,6 +235,45 @@ def test_create_primary_key_blob_tables(self):
235235
AtomicType("BLOB"),
236236
{"blob-view-field": "payload", "changelog-producer": "input"},
237237
),
238+
(
239+
"partial_update_ignore_retract",
240+
AtomicType("BLOB"),
241+
{
242+
"merge-engine": "partial-update",
243+
"fields.ts.sequence-group": "payload",
244+
"fields.payload.aggregate-function": "last_non_null_value",
245+
"fields.payload.ignore-retract": "true",
246+
},
247+
),
248+
(
249+
"partial_update_ignore_delete",
250+
AtomicType("BLOB"),
251+
{
252+
"merge-engine": "partial-update",
253+
"fields.ts.sequence-group": "payload",
254+
"fields.payload.aggregate-function": "last_non_null_value",
255+
"partial-update.ignore-delete": "true",
256+
},
257+
),
258+
(
259+
"partial_update_last_value",
260+
AtomicType("BLOB"),
261+
{
262+
"merge-engine": "partial-update",
263+
"fields.ts.sequence-group": "payload",
264+
"fields.payload.aggregate-function": "last_value",
265+
},
266+
),
267+
(
268+
"partial_update_inline_blob_agg",
269+
AtomicType("BLOB"),
270+
{
271+
"merge-engine": "partial-update",
272+
"blob-descriptor-field": "payload",
273+
"fields.ts.sequence-group": "payload",
274+
"fields.payload.aggregate-function": "last_non_null_value",
275+
},
276+
),
238277
]
239278
for suffix, blob_type, extra_options in blob_types:
240279
with self.subTest(kind=suffix):
@@ -245,14 +284,15 @@ def test_create_primary_key_blob_tables(self):
245284
schema = Schema(
246285
fields=[
247286
DataField(0, "id", AtomicType("INT", False)),
248-
DataField(1, "payload", blob_type),
287+
DataField(1, "ts", AtomicType("INT", False)),
288+
DataField(2, "payload", blob_type),
249289
],
250290
primary_keys=["id"],
251291
options=extra_options,
252292
)
253293
table_schema = manager.create_table(schema)
254294
self.assertEqual(table_schema.primary_keys, ["id"])
255-
self.assertEqual(table_schema.fields[1].type, blob_type)
295+
self.assertEqual(table_schema.fields[2].type, blob_type)
256296

257297
def test_create_primary_key_blob_table_validation(self):
258298
cases = [
@@ -370,6 +410,103 @@ def test_create_primary_key_blob_table_validation(self):
370410
{"pk-clustering-override": "true"},
371411
"Primary-key managed BLOB tables do not support 'pk-clustering-override'.",
372412
),
413+
(
414+
"managed_blob_partial_update_retract_agg",
415+
[
416+
DataField(0, "id", AtomicType("INT", False)),
417+
DataField(1, "ts", AtomicType("INT", False)),
418+
DataField(2, "payload", AtomicType("BLOB")),
419+
],
420+
["id"],
421+
{
422+
"merge-engine": "partial-update",
423+
"fields.ts.sequence-group": "payload",
424+
"fields.payload.aggregate-function": "last_non_null_value",
425+
},
426+
r"Managed BLOB field 'payload' cannot use aggregate function 'last_non_null_value' "
427+
r"because managed BLOB payloads are not retained in retract messages\. "
428+
r"Set 'fields\.payload\.ignore-retract' to true to ignore retract messages\.",
429+
),
430+
(
431+
"managed_blob_partial_update_default_agg",
432+
[
433+
DataField(0, "id", AtomicType("INT", False)),
434+
DataField(1, "ts", AtomicType("INT", False)),
435+
DataField(2, "payload", AtomicType("BLOB")),
436+
],
437+
["id"],
438+
{
439+
"merge-engine": "partial-update",
440+
"fields.ts.sequence-group": "payload",
441+
"fields.default-aggregate-function": "last_non_null_value",
442+
},
443+
r"Managed BLOB field 'payload' cannot use aggregate function 'last_non_null_value' "
444+
r"because managed BLOB payloads are not retained in retract messages\. "
445+
r"Set 'fields\.payload\.ignore-retract' to true to ignore retract messages\.",
446+
),
447+
(
448+
"managed_blob_as_sequence_group_ordering_field",
449+
[
450+
DataField(0, "id", AtomicType("INT", False)),
451+
DataField(1, "ts", AtomicType("INT", False)),
452+
DataField(2, "payload", AtomicType("BLOB")),
453+
],
454+
["id"],
455+
{
456+
"merge-engine": "partial-update",
457+
"fields.payload.sequence-group": "ts",
458+
},
459+
r"Field 'payload' with type BLOB cannot be used as a sequence-group ordering "
460+
r"field in option 'fields\.payload\.sequence-group'\.",
461+
),
462+
(
463+
"inline_blob_as_sequence_group_ordering_field",
464+
[
465+
DataField(0, "id", AtomicType("INT", False)),
466+
DataField(1, "ts", AtomicType("INT", False)),
467+
DataField(2, "payload", AtomicType("BLOB")),
468+
],
469+
["id"],
470+
{
471+
"merge-engine": "partial-update",
472+
"blob-descriptor-field": "payload",
473+
"fields.payload.sequence-group": "ts",
474+
},
475+
r"Field 'payload' with type BLOB cannot be used as a sequence-group ordering "
476+
r"field in option 'fields\.payload\.sequence-group'\.",
477+
),
478+
(
479+
"pk_as_sequence_group_ordering_field",
480+
[
481+
DataField(0, "id", AtomicType("INT", False)),
482+
DataField(1, "ts", AtomicType("INT", False)),
483+
DataField(2, "payload", AtomicType("BLOB")),
484+
],
485+
["id"],
486+
{
487+
"merge-engine": "partial-update",
488+
"fields.id.sequence-group": "payload",
489+
},
490+
r"The sequence-group 'fields\.id\.sequence-group' contains primary key field "
491+
r"'id', which is not allowed\. Primary key columns cannot be put in "
492+
r"sequence-group\.",
493+
),
494+
(
495+
"pk_as_sequence_group_protected_field",
496+
[
497+
DataField(0, "id", AtomicType("INT", False)),
498+
DataField(1, "ts", AtomicType("INT", False)),
499+
DataField(2, "payload", AtomicType("BLOB")),
500+
],
501+
["id"],
502+
{
503+
"merge-engine": "partial-update",
504+
"fields.ts.sequence-group": "id,payload",
505+
},
506+
r"The sequence-group 'fields\.ts\.sequence-group' contains primary key field "
507+
r"'id', which is not allowed\. Primary key columns cannot be put in "
508+
r"sequence-group\.",
509+
),
373510
]
374511

375512
for name, fields, pks, opts, error_regex in cases:

0 commit comments

Comments
 (0)