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 fb02b3cf01 [core][spark][flink][python] Support ignoring global index
column updates (#8793)
fb02b3cf01 is described below
commit fb02b3cf0170f5b5ab41fe25bbe4ad4d0b8a6888
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 22 14:39:12 2026 +0800
[core][spark][flink][python] Support ignoring global index column updates
(#8793)
Allow data-evolution updates to proceed without invalidating existing
global index files when the caller explicitly accepts temporary index
staleness. One example is updating a vector column from `NULL` to a
value: existing indexed results remain valid, while the new vector stays
invisible until the index is rebuilt.
---
docs/docs/multimodal-table/global-index.mdx | 1 +
docs/generated/core_configuration.html | 2 +-
.../main/java/org/apache/paimon/CoreOptions.java | 8 +++++--
.../java/org/apache/paimon/CoreOptionsTest.java | 9 +++++++
.../dataevolution/MergeIntoUpdateChecker.java | 8 +++++--
.../action/DataEvolutionMergeIntoActionITCase.java | 19 +++++++++++++--
.../pypaimon/common/options/core_options.py | 4 +++-
.../tests/global_index_update_action_test.py | 19 +++++++++++++++
.../pypaimon/write/global_index_update_checker.py | 4 +++-
.../MergeIntoPaimonDataEvolutionTable.scala | 6 +++++
.../MergeIntoPaimonDataEvolutionTable.scala | 6 +++++
.../paimon/spark/sql/RowTrackingTestBase.scala | 28 ++++++++++++++++++++++
12 files changed, 105 insertions(+), 9 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index a09a76e962..ace4770ba7 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -391,6 +391,7 @@ These table options affect global index build and read
behavior:
| `global-index.enabled` | `true` | Whether scans can use global indexes. |
| `global-index.search-mode` | `fast` | Search mode for global-index queries.
`fast` searches indexed data only. `full` checks snapshot `nextRowId` against
global index row-id coverage and scans raw data only if a gap exists. `detail`
scans data file metadata to find exact unindexed rows and can handle index
invalidation caused by updates or rewrites. |
| `global-index.external-path` | Not set | Root directory for global index
files. If not set, files are stored under the table index directory. |
+| `global-index.column-update-action` | `THROW_ERROR` | Action for updates to
indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops
affected partition indexes, and `IGNORE` keeps existing index files unchanged.
Use `IGNORE` only when a stale index is acceptable; for example, a vector
updated from `NULL` to a value remains invisible until the index is rebuilt. |
| `sorted-index.records-per-range` | `10000000` | Expected number of records
per sorted global index file for BTree and Bitmap builds. |
| `sorted-index.build.max-parallelism` | `4096` | Maximum Flink or Spark
parallelism for building sorted global indexes. |
| `global-index.row-count-per-shard` | `100000` | Target row count per shard
for non-sorted global index builds such as vector and full-text indexes. |
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 6c077fa9e4..0776eaf9e3 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -768,7 +768,7 @@ under the License.
<td><h5>global-index.column-update-action</h5></td>
<td style="word-wrap: break-word;">THROW_ERROR</td>
<td><p>Enum</p></td>
- <td>Defines the action to take when an update modifies columns
that are covered by a global index.<br /><br />Possible
values:<ul><li>"THROW_ERROR"</li><li>"DROP_PARTITION_INDEX"</li></ul></td>
+ <td>Defines the action to take when an update modifies columns
that are covered by a global index. IGNORE leaves existing index files
unchanged and may make the index stale.<br /><br />Possible
values:<ul><li>"THROW_ERROR"</li><li>"DROP_PARTITION_INDEX"</li><li>"IGNORE"</li></ul></td>
</tr>
<tr>
<td><h5>global-index.enabled</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 6f6869eb18..ec6a2296bf 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2504,7 +2504,8 @@ public class CoreOptions implements Serializable {
.enumType(GlobalIndexColumnUpdateAction.class)
.defaultValue(GlobalIndexColumnUpdateAction.THROW_ERROR)
.withDescription(
- "Defines the action to take when an update
modifies columns that are covered by a global index.");
+ "Defines the action to take when an update
modifies columns that are covered by a global index. "
+ + "IGNORE leaves existing index
files unchanged and may make the index stale.");
public static final ConfigOption<MemorySize> LOOKUP_MERGE_BUFFER_SIZE =
key("lookup.merge-buffer-size")
@@ -5406,7 +5407,10 @@ public class CoreOptions implements Serializable {
THROW_ERROR,
/** Drop all global index entries for the whole partitions affected by
the update. */
- DROP_PARTITION_INDEX
+ DROP_PARTITION_INDEX,
+
+ /** Leave existing global index entries unchanged when indexed columns
are updated. */
+ IGNORE
}
/** Search mode for global index queries. */
diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
index a996942292..b3e91a3a4b 100644
--- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java
@@ -98,6 +98,15 @@ public class CoreOptionsTest {
assertThat(options.sequenceField()).containsExactly("f1", "f2", "f3");
}
+ @Test
+ public void testIgnoreGlobalIndexColumnUpdateAction() {
+ Options conf = new Options();
+ conf.setString(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
"IGNORE");
+
+ assertThat(new CoreOptions(conf).globalIndexColumnUpdateAction())
+ .isEqualTo(CoreOptions.GlobalIndexColumnUpdateAction.IGNORE);
+ }
+
@Test
public void testBlobSplitByFileSizeDefault() {
Options conf = new Options();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
index bdd0c0d491..0f749c18bd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/MergeIntoUpdateChecker.java
@@ -87,6 +87,12 @@ public class MergeIntoUpdateChecker extends
BoundedOneInputOperator<Committable,
}
private void checkUpdatedColumns() {
+ CoreOptions.GlobalIndexColumnUpdateAction updateAction =
+ table.coreOptions().globalIndexColumnUpdateAction();
+ if (updateAction == CoreOptions.GlobalIndexColumnUpdateAction.IGNORE) {
+ return;
+ }
+
Optional<Snapshot> latestSnapshot = table.latestSnapshot();
RowType rowType = table.rowType();
Preconditions.checkState(latestSnapshot.isPresent());
@@ -112,8 +118,6 @@ public class MergeIntoUpdateChecker extends
BoundedOneInputOperator<Committable,
});
if (!affectedEntries.isEmpty()) {
- CoreOptions.GlobalIndexColumnUpdateAction updateAction =
- table.coreOptions().globalIndexColumnUpdateAction();
switch (updateAction) {
case THROW_ERROR:
Set<String> conflictedColumns =
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
index b944a07a5c..2deff25d13 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java
@@ -561,6 +561,21 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
.hasMessageContaining(
"MergeInto: update columns contain globally indexed
columns, not supported now.");
+ // 2. IGNORE should allow the update and leave the index unchanged
+ executeSQL(
+ "ALTER TABLE T SET ('global-index.column-update-action' =
'IGNORE')", false, true);
+
+ assertDoesNotThrow(
+ () ->
+ executeSQL(
+ String.format(
+ "CALL
sys.data_evolution_merge_into('%s.T', '', '', 'S', 'T._ROW_ID=S.id',
'name=S.name,id=1', 2)",
+ database),
+ false,
+ true));
+
+ assertTrue(indexFileExists("T"));
+
insertInto(
"T",
"(31, 'name31', 3.1, '01-23')",
@@ -576,7 +591,7 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
insertInto("S", "(35, 'new_name25', 125.1)");
- // 2. updating unindexed partitions is not affected
+ // 3. updating unindexed partitions is not affected
assertDoesNotThrow(
() ->
executeSQL(
@@ -588,7 +603,7 @@ public class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
false,
true));
- // 3. alter table's UpdateAction option to DROP_INDEX
+ // 4. alter table's UpdateAction option to DROP_INDEX
executeSQL(
"ALTER TABLE T SET ('global-index.column-update-action' =
'DROP_PARTITION_INDEX')",
false,
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 36fd975f6d..3b890cc0d6 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -90,6 +90,7 @@ class StartupMode(str, Enum):
class GlobalIndexColumnUpdateAction(str, Enum):
THROW_ERROR = "THROW_ERROR"
DROP_PARTITION_INDEX = "DROP_PARTITION_INDEX"
+ IGNORE = "IGNORE"
class GlobalIndexSearchMode(str, Enum):
@@ -787,7 +788,8 @@ class CoreOptions:
.default_value(GlobalIndexColumnUpdateAction.THROW_ERROR)
.with_description(
"Defines the action to take when an update modifies columns that "
- "are covered by a global index."
+ "are covered by a global index. IGNORE leaves existing index files
"
+ "unchanged and may make the index stale."
)
)
diff --git a/paimon-python/pypaimon/tests/global_index_update_action_test.py
b/paimon-python/pypaimon/tests/global_index_update_action_test.py
index 8cd92e253e..17ddf67eb0 100644
--- a/paimon-python/pypaimon/tests/global_index_update_action_test.py
+++ b/paimon-python/pypaimon/tests/global_index_update_action_test.py
@@ -126,6 +126,25 @@ class GlobalIndexUpdateActionTest(unittest.TestCase):
self.assertIn("'age'", str(ctx.exception))
self.assertIn("Conflicted columns: ['age']", str(ctx.exception))
+ def test_ignore_action_skips_global_index_handling(self):
+ options = CoreOptions.from_dict({
+ CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(): "IGNORE",
+ })
+ table = _Table(options)
+
+ with mock.patch(
+ "pypaimon.write.global_index_update_checker."
+ "scan_global_index_entries") as scan:
+ messages = apply_global_index_update_action(
+ table,
+ object(),
+ ["name"],
+ {()},
+ )
+
+ self.assertEqual([], messages)
+ scan.assert_not_called()
+
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",
diff --git a/paimon-python/pypaimon/write/global_index_update_checker.py
b/paimon-python/pypaimon/write/global_index_update_checker.py
index db44e5e4af..b7f270c17c 100644
--- a/paimon-python/pypaimon/write/global_index_update_checker.py
+++ b/paimon-python/pypaimon/write/global_index_update_checker.py
@@ -66,6 +66,9 @@ def apply_global_index_update_action(
) -> list:
if snapshot is None or not updated_cols or not written_partitions:
return []
+ action = table.options.global_index_column_update_action()
+ if action == GlobalIndexColumnUpdateAction.IGNORE:
+ return []
entries = scan_global_index_entries(table, snapshot)
if not entries:
return []
@@ -84,7 +87,6 @@ def apply_global_index_update_action(
conflicted.update(matched)
if not affected:
return []
- action = table.options.global_index_column_update_action()
if action is None:
action = GlobalIndexColumnUpdateAction.THROW_ERROR
if action == GlobalIndexColumnUpdateAction.DROP_PARTITION_INDEX:
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 1619dfd976..4e4b21f4e2 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -926,6 +926,12 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def checkUpdateResult(updateCommit: Seq[CommitMessage]):
Seq[CommitMessage] = {
+ if (
+ table.coreOptions().globalIndexColumnUpdateAction() ==
GlobalIndexColumnUpdateAction.IGNORE
+ ) {
+ return updateCommit
+ }
+
val affectedParts: Set[BinaryRow] = updateCommit.map(_.partition()).toSet
val rowType = table.rowType()
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 1619dfd976..4e4b21f4e2 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -926,6 +926,12 @@ case class MergeIntoPaimonDataEvolutionTable(
}
private def checkUpdateResult(updateCommit: Seq[CommitMessage]):
Seq[CommitMessage] = {
+ if (
+ table.coreOptions().globalIndexColumnUpdateAction() ==
GlobalIndexColumnUpdateAction.IGNORE
+ ) {
+ return updateCommit
+ }
+
val affectedParts: Set[BinaryRow] = updateCommit.map(_.partition()).toSet
val rowType = table.rowType()
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index dc96e3ab1e..72258ddbec 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -1381,4 +1381,32 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: test global indexed column update action -- ignore") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, name STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'global-index.column-update-action' = 'IGNORE')
+ |""".stripMargin)
+ sql("INSERT INTO T VALUES (1, 'name_1')")
+ sql(
+ "CALL sys.create_global_index(table => 'test.T', index_column =>
'name', " +
+ "index_type => 'btree')")
+
+ sql("""
+ |MERGE INTO T
+ |USING T AS source
+ |ON T._ROW_ID = source._ROW_ID
+ |WHEN MATCHED THEN UPDATE SET name = 'updated_name'
+ |""".stripMargin)
+
+ checkAnswer(sql("SELECT id, name FROM T"), Seq(Row(1, "updated_name")))
+ val indexEntries =
loadTable("T").store().newIndexFileHandler().scan("btree")
+ assert(!indexEntries.isEmpty)
+ }
+ }
+
}
