This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 36caecf49e [python] Support global index column update action (#8270)
36caecf49e is described below
commit 36caecf49e0d1690005df5db5d20f68b94afda62
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jun 17 23:02:36 2026 +0800
[python] Support global index column update action (#8270)
Support Python API usage of `global-index.column-update-action` for
global index column updates. This keeps the Python option behavior
aligned with the Java enum-style configuration and documents the option
in Python core options.
---
.../pypaimon/common/options/core_options.py | 4 +
.../pypaimon/common/options/options_utils.py | 2 +
.../tests/global_index_update_action_test.py | 191 +++++++++++++++++++++
.../pypaimon/write/global_index_update_checker.py | 32 +++-
4 files changed, 220 insertions(+), 9 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index e2df232193..78c8db22e5 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -632,6 +632,10 @@ class CoreOptions:
ConfigOptions.key("global-index.column-update-action")
.enum_type(GlobalIndexColumnUpdateAction)
.default_value(GlobalIndexColumnUpdateAction.THROW_ERROR)
+ .with_description(
+ "Defines the action to take when an update modifies columns that "
+ "are covered by a global index."
+ )
)
LOCAL_CACHE_ENABLED: ConfigOption[bool] = (
diff --git a/paimon-python/pypaimon/common/options/options_utils.py
b/paimon-python/pypaimon/common/options/options_utils.py
index 7933abe056..4c86e7898d 100644
--- a/paimon-python/pypaimon/common/options/options_utils.py
+++ b/paimon-python/pypaimon/common/options/options_utils.py
@@ -72,6 +72,8 @@ class OptionsUtils:
@staticmethod
def convert_to_string(value: Any) -> str:
"""Convert value to string."""
+ if isinstance(value, Enum):
+ return str(value.value)
return str(value)
@staticmethod
diff --git a/paimon-python/pypaimon/tests/global_index_update_action_test.py
b/paimon-python/pypaimon/tests/global_index_update_action_test.py
new file mode 100644
index 0000000000..8cd92e253e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/global_index_update_action_test.py
@@ -0,0 +1,191 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from unittest import mock
+
+from pypaimon.common.options.core_options import (
+ CoreOptions,
+ GlobalIndexColumnUpdateAction,
+)
+from pypaimon.common.options.options import Options
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.global_index_update_checker import (
+ apply_global_index_update_action,
+)
+
+
+def _field(field_id, name, field_type="STRING"):
+ return DataField(field_id, name, AtomicType(field_type))
+
+
+def _partition_row(values):
+ fields = [_field(i, "p{}".format(i)) for i in range(len(values))]
+ return GenericRow(list(values), fields)
+
+
+def _index_entry(file_name, partition, field_id, extra_field_ids=None):
+ index_file = IndexFileMeta(
+ index_type="BTREE",
+ file_name=file_name,
+ file_size=1,
+ row_count=1,
+ global_index_meta=GlobalIndexMeta(
+ row_range_start=0,
+ row_range_end=0,
+ index_field_id=field_id,
+ extra_field_ids=extra_field_ids,
+ index_meta=b"",
+ ),
+ )
+ return IndexManifestEntry(
+ kind=0,
+ partition=_partition_row(partition),
+ bucket=0,
+ index_file=index_file,
+ )
+
+
+class _Table:
+ fields = [_field(1, "name"), _field(2, "age", "INT")]
+
+ def __init__(self, options):
+ self.options = options
+
+
+class GlobalIndexUpdateActionTest(unittest.TestCase):
+
+ def test_column_update_action_reads_java_style_string(self):
+ options = CoreOptions.from_dict({
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key():
"DROP_PARTITION_INDEX",
+ })
+
+ self.assertEqual(
+ GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX,
+ options.global_index_column_update_action(),
+ )
+
+ def test_column_update_action_round_trips_python_enum(self):
+ options = CoreOptions(Options({}))
+ options.set(
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION,
+ GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX,
+ )
+
+ self.assertEqual(
+ "DROP_PARTITION_INDEX",
+ options.options.to_map()[
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key()
+ ],
+ )
+ self.assertEqual(
+ GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX,
+ options.global_index_column_update_action(),
+ )
+
+ def test_default_action_rejects_updates_to_global_index_columns(self):
+ table = _Table(CoreOptions.from_dict({}))
+ entries = [_index_entry("idx-name", (), 1)]
+
+ with mock.patch(
+ "pypaimon.write.global_index_update_checker."
+ "scan_global_index_entries",
+ return_value=entries):
+ with self.assertRaisesRegex(RuntimeError, "Conflicted columns"):
+ apply_global_index_update_action(table, object(), ["name"],
{()})
+
+ def test_default_action_rejects_extra_global_index_columns(self):
+ table = _Table(CoreOptions.from_dict({}))
+ entries = [_index_entry("idx-name-age", (), 1, extra_field_ids=[2])]
+
+ with mock.patch(
+ "pypaimon.write.global_index_update_checker."
+ "scan_global_index_entries",
+ return_value=entries):
+ with self.assertRaises(RuntimeError) as ctx:
+ apply_global_index_update_action(table, object(), ["age"],
{()})
+
+ self.assertIn("'age'", str(ctx.exception))
+ self.assertIn("Conflicted columns: ['age']", str(ctx.exception))
+
+ def test_drop_partition_index_builds_deletes_for_affected_partition(self):
+ options = CoreOptions.from_dict({
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key():
"DROP_PARTITION_INDEX",
+ })
+ table = _Table(options)
+ entries = [
+ _index_entry("idx-name-p0", ("2026-06-17",), 1),
+ _index_entry("idx-age-p0", ("2026-06-17",), 2),
+ _index_entry("idx-name-p1", ("2026-06-18",), 1),
+ ]
+
+ with mock.patch(
+ "pypaimon.write.global_index_update_checker."
+ "scan_global_index_entries",
+ return_value=entries):
+ messages = apply_global_index_update_action(
+ table,
+ object(),
+ ["name"],
+ {("2026-06-17",)},
+ )
+
+ self.assertEqual(1, len(messages))
+ self.assertEqual(("2026-06-17",), messages[0].partition)
+ self.assertEqual(
+ ["idx-name-p0"],
+ [d.index_file.file_name for d in messages[0].index_deletes],
+ )
+ self.assertEqual([1], [d.kind for d in messages[0].index_deletes])
+
+ def test_drop_partition_index_builds_deletes_for_extra_column_update(self):
+ options = CoreOptions.from_dict({
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key():
"DROP_PARTITION_INDEX",
+ })
+ table = _Table(options)
+ entries = [
+ _index_entry(
+ "idx-name-age-p0",
+ ("2026-06-17",),
+ 1,
+ extra_field_ids=[2],
+ ),
+ ]
+
+ with mock.patch(
+ "pypaimon.write.global_index_update_checker."
+ "scan_global_index_entries",
+ return_value=entries):
+ messages = apply_global_index_update_action(
+ table,
+ object(),
+ ["age"],
+ {("2026-06-17",)},
+ )
+
+ self.assertEqual(
+ ["idx-name-age-p0"],
+ [d.index_file.file_name for d in messages[0].index_deletes],
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/write/global_index_update_checker.py
b/paimon-python/pypaimon/write/global_index_update_checker.py
index ac405fc4bc..db44e5e4af 100644
--- a/paimon-python/pypaimon/write/global_index_update_checker.py
+++ b/paimon-python/pypaimon/write/global_index_update_checker.py
@@ -47,6 +47,17 @@ def build_index_delete_msgs(entries) -> list:
]
+def indexed_field_names(global_index_meta, field_by_id) -> set:
+ field_ids = [global_index_meta.index_field_id]
+ if global_index_meta.extra_field_ids:
+ field_ids.extend(global_index_meta.extra_field_ids)
+ return {
+ field_by_id[field_id]
+ for field_id in field_ids
+ if field_id in field_by_id
+ }
+
+
def apply_global_index_update_action(
table,
snapshot,
@@ -60,11 +71,17 @@ def apply_global_index_update_action(
return []
field_by_id = {f.id: f.name for f in table.fields}
update_set = set(updated_cols)
- affected = [
- e for e in entries
- if field_by_id.get(e.index_file.global_index_meta.index_field_id) in
update_set
- and tuple(e.partition.values) in written_partitions
- ]
+ affected = []
+ conflicted = set()
+ for e in entries:
+ if tuple(e.partition.values) not in written_partitions:
+ continue
+ matched = indexed_field_names(
+ e.index_file.global_index_meta, field_by_id
+ ).intersection(update_set)
+ if matched:
+ affected.append(e)
+ conflicted.update(matched)
if not affected:
return []
action = table.options.global_index_column_update_action()
@@ -72,11 +89,8 @@ def apply_global_index_update_action(
action = GlobalIndexColumnUpdateAction.THROW_ERROR
if action == GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX:
return build_index_delete_msgs(affected)
- conflicted = sorted(
- {field_by_id.get(e.index_file.global_index_meta.index_field_id) for e
in affected}
- )
raise RuntimeError(
f"Update columns contain globally indexed columns, not supported
now.\n"
f"Updated columns: {sorted(update_set)}\n"
- f"Conflicted columns: {conflicted}"
+ f"Conflicted columns: {sorted(conflicted)}"
)
