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 d76769838b [python] Align INT/BIGINT and timestamp evolution with Java
(#10508)
d76769838b is described below
commit d76769838bd1c191e8e62167f9f173c66a252c21
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 12 13:55:18 2026 +0800
[python] Align INT/BIGINT and timestamp evolution with Java (#10508)
---
paimon-python/pypaimon/casting/data_type_casts.py | 15 ++-
paimon-python/pypaimon/casting/schema_evolution.py | 46 +++++++-
.../pypaimon/tests/native_predicate_delete_test.py | 2 +-
.../tests/native_schema_evolution_cast_test.py | 121 +++++++++++++++++++++
.../pypaimon/tests/schema_evolution_cast_test.py | 58 ++++++++++
.../tests/schema_evolution_nested_read_test.py | 9 ++
paimon-python/pypaimon/tests/table_update_test.py | 36 +++++-
paimon-python/pypaimon/write/table_delete.py | 8 +-
8 files changed, 280 insertions(+), 15 deletions(-)
diff --git a/paimon-python/pypaimon/casting/data_type_casts.py
b/paimon-python/pypaimon/casting/data_type_casts.py
index 819b9c115c..849be57e54 100644
--- a/paimon-python/pypaimon/casting/data_type_casts.py
+++ b/paimon-python/pypaimon/casting/data_type_casts.py
@@ -192,7 +192,7 @@ def _equals_ignore_nullable(source_type, target_type) ->
bool:
return source_copy == target_copy
-# Caches the PyArrow cast-kernel probe per (source, target) pyarrow type so the
+# Caches the read-time cast probe per (source, target) pyarrow type so the
# alter-time check stays cheap. Keyed by the pyarrow type strings.
_EXECUTABLE_CAST_CACHE = {}
@@ -225,12 +225,14 @@ def can_execute_cast(source_type, target_type) -> bool:
# Any other conversion touching a constructed type has no runtime cast.
if source_root in CONSTRUCTED or target_root in CONSTRUCTED:
return False
- # Leaf-to-leaf: defer to PyArrow's cast-kernel availability, which is the
- # read path's actual cast executor (``array.cast(target, safe=False)``).
- return _pyarrow_cast_supported(source_type, target_type)
+ # Leaf-to-leaf: probe the same executor used by old-file readers, including
+ # Java conversions with no direct PyArrow kernel, such as INT -> TIMESTAMP.
+ return _schema_evolution_cast_supported(source_type, target_type)
-def _pyarrow_cast_supported(source_type, target_type) -> bool:
+def _schema_evolution_cast_supported(source_type, target_type) -> bool:
+ from pypaimon.casting.schema_evolution import
cast_array_for_schema_evolution
+
source_pa = PyarrowFieldParser.from_paimon_type(source_type)
target_pa = PyarrowFieldParser.from_paimon_type(target_type)
if source_pa == target_pa:
@@ -248,7 +250,8 @@ def _pyarrow_cast_supported(source_type, target_type) ->
bool:
# avoiding per-value parse/overflow errors (which ``safe=False`` --
matching
# the read path -- tolerates anyway).
try:
- pa.nulls(1, type=source_pa).cast(target_pa, safe=False)
+ cast_array_for_schema_evolution(
+ pa.nulls(1, type=source_pa), target_pa, source_type, target_type)
ok = True
except (pa.lib.ArrowNotImplementedError, pa.lib.ArrowInvalid,
pa.lib.ArrowTypeError):
diff --git a/paimon-python/pypaimon/casting/schema_evolution.py
b/paimon-python/pypaimon/casting/schema_evolution.py
index 62793ec79c..b01cc99d7e 100644
--- a/paimon-python/pypaimon/casting/schema_evolution.py
+++ b/paimon-python/pypaimon/casting/schema_evolution.py
@@ -77,12 +77,49 @@ def _truncate_timestamp(array, target_type,
source_paimon_type, target_paimon_ty
return pa.array(values, type=pa.int64()).cast(target_type)
+def _timestamp_to_bigint(array):
+ ticks = {'s': 1, 'ms': 1000, 'us': 1000000, 'ns':
1000000000}[array.type.unit]
+ values = []
+ for value in array.cast(pa.int64()).to_pylist():
+ if value is None:
+ values.append(None)
+ elif array.type.tz is None:
+ values.append(value // ticks)
+ else:
+ # Java's LTZ rule round-trips Timestamp through LocalDateTime.
+ # Timestamp stores its epoch day as an int in that conversion.
+ millis = value * 1000 if array.type.unit == 's' else value //
(ticks // 1000)
+ if not -(1 << 63) <= millis < (1 << 63):
+ values.append(None)
+ continue
+ day = _narrow_integer(millis // 86400000, 32)
+ values.append((day * 86400000 + millis % 86400000) // 1000)
+ return pa.array(values, type=pa.int64())
+
+
+def _integer_to_timestamp(array, target_type):
+ ticks = {'s': 1, 'ms': 1000, 'us': 1000000, 'ns':
1000000000}[target_type.unit]
+ values = []
+ for value in array.to_pylist():
+ if value is None:
+ values.append(None)
+ continue
+ # NumericPrimitiveToTimestamp performs value.longValue() * 1000
+ # before conversion. Java's signed long multiplication wraps.
+ millis = _narrow_integer(value * 1000, 64)
+ value = millis // 1000 if target_type.unit == 's' else millis * (ticks
// 1000)
+ values.append(value if -(1 << 63) <= value < (1 << 63) else None)
+ return pa.array(values, type=pa.int64()).cast(target_type)
+
+
def cast_array_for_schema_evolution(array, target_type,
source_paimon_type=None, target_paimon_type=None):
"""Materialize existing type changes using Java numeric and timestamp
casts.
Declared Paimon types are needed for precision changes within one Arrow
- timestamp unit. Numeric narrowing preserves low bits, while floating-point
- values truncate and saturate to int/long before byte/short narrowing.
+ timestamp unit. INTEGER/BIGINT to TIMESTAMP and TIMESTAMP to BIGINT use
+ epoch seconds, including pre-epoch flooring and Java's millis
multiplication.
+ Numeric narrowing preserves low bits, while floating-point values truncate
+ and saturate to int/long before byte/short narrowing.
DECIMAL scale reduction uses HALF_UP and returns NULL on precision
overflow.
"""
timestamp = _truncate_timestamp(array, target_type, source_paimon_type,
target_paimon_type)
@@ -90,6 +127,11 @@ def cast_array_for_schema_evolution(array, target_type,
source_paimon_type=None,
return timestamp
if array.type == target_type:
return array
+ if pa.types.is_timestamp(array.type) and target_type == pa.int64():
+ return _timestamp_to_bigint(array)
+ if (array.type in (pa.int32(), pa.int64())
+ and pa.types.is_timestamp(target_type) and target_type.tz is None):
+ return _integer_to_timestamp(array, target_type)
if (pa.types.is_signed_integer(target_type)
and (pa.types.is_floating(array.type) or
pa.types.is_decimal(array.type))):
return _numeric_to_integer(array, target_type)
diff --git a/paimon-python/pypaimon/tests/native_predicate_delete_test.py
b/paimon-python/pypaimon/tests/native_predicate_delete_test.py
index 1484b102f6..4622dab0af 100644
--- a/paimon-python/pypaimon/tests/native_predicate_delete_test.py
+++ b/paimon-python/pypaimon/tests/native_predicate_delete_test.py
@@ -176,7 +176,7 @@ def
test_data_delete_rejects_disabled_dv_without_fallback(native_rest_catalog, m
predicate = pb.equal('id', 999)
if mixed:
predicate = pb.or_predicates([predicate, pb.equal('p', 'a')])
- with _core_only(), pytest.raises(Exception,
match='deletion-vectors.enabled'):
+ with _core_only(), pytest.raises(NotImplementedError,
match='deletion-vectors.enabled'):
table.new_batch_write_builder().new_update().delete_by_predicate(predicate)
assert _ids(table) == [0, 1, 2, 3]
diff --git a/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
index 4f6d7b6400..f910737817 100644
--- a/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
+++ b/paimon-python/pypaimon/tests/native_schema_evolution_cast_test.py
@@ -214,3 +214,124 @@ def
test_nested_narrowing_preserves_null_containers_and_leaf_projection(rest_cat
leaves = _read(table, native, projection=['id', 'r.n', 'r.t'])
assert leaves.column('r_n').to_pylist() == [-128, None, 127]
assert leaves.column('r_t').cast(pa.int64()).to_pylist() == [100,
None, -100]
+
+
+_TEMPORAL_CASES = [
+ (pa.timestamp('ms'), 'BIGINT', [42001, -1, -42001, 2147483650000, None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.timestamp('us'), 'BIGINT', [42000001, -1, -42000001, 2147483650000000,
None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.timestamp('ns'), 'BIGINT', [42000000001, -1, -42000000001,
2147483650000000000, None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.timestamp('ms', 'UTC'), 'BIGINT', [42001, -1, -42001, 2147483650000,
None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.timestamp('us', 'UTC'), 'BIGINT', [42000001, -1, -42000001,
2147483650000000, None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.timestamp('ns', 'UTC'), 'BIGINT', [42000000001, -1, -42000000001,
2147483650000000000, None],
+ [42, -1, -43, 2147483650, None]),
+ (pa.int32(), 'TIMESTAMP(0)', [42, -42, 2147483647, -2147483648, None],
+ [42, -42, 2147483647, -2147483648, None]),
+ (pa.int32(), 'TIMESTAMP(1)', [42, -42, 2147483647, -2147483648, None],
+ [42000, -42000, 2147483647000, -2147483648000, None]),
+ (pa.int32(), 'TIMESTAMP(4)', [42, -42, 2147483647, -2147483648, None],
+ [42000000, -42000000, 2147483647000000, -2147483648000000, None]),
+ (pa.int32(), 'TIMESTAMP(7)', [42, -42, 2147483647, -2147483648, None],
+ [42000000000, -42000000000, 2147483647000000000, -2147483648000000000,
None]),
+ (pa.int64(), 'TIMESTAMP(0)', [42, -42, 2147483650, 2 ** 63 - 1, -(2 **
63), None],
+ [42, -42, 2147483650, -1, 0, None]),
+ (pa.int64(), 'TIMESTAMP(1)', [42, -42, 2147483650, 2 ** 63 - 1, -(2 **
63), None],
+ [42000, -42000, 2147483650000, -1000, 0, None]),
+ (pa.int64(), 'TIMESTAMP(4)', [42, -42, 2147483650, 2 ** 63 - 1, -(2 **
63), None],
+ [42000000, -42000000, 2147483650000000, -1000000, 0, None]),
+ (pa.int64(), 'TIMESTAMP(7)', [42, -42, 2147483650, 2 ** 63 - 1, -(2 **
63), None],
+ [42000000000, -42000000000, 2147483650000000000, -1000000000, 0, None]),
+]
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('source_type,target_name,values,expected',
_TEMPORAL_CASES)
+def test_epoch_seconds_evolution_reads_old_and_new_rest_files(
+ rest_catalog, mode, source_type, target_name, values, expected):
+ catalog, _ = rest_catalog
+ _table(catalog, mode, source_type, values)
+ target = AtomicType(target_name)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
target)], False)
+ table = catalog.get_table('default.t')
+ target_type = PyarrowFieldParser.from_paimon_type(target)
+ for native in (False, True):
+ result = _read(table, native).column('v')
+ assert result.type == target_type
+ assert result.cast(pa.int64()).to_pylist() == expected
+ # Both schema versions must materialize the current type in one read.
+ _write(table, pa.table({'id': [len(values)], 'v': pa.array([expected[0]],
type=target_type)}))
+ table = catalog.get_table('default.t')
+ first = pa.scalar(expected[0], type=target_type).as_py()
+ pb = table.new_read_builder().new_predicate_builder()
+ for native in (False, True):
+ assert _read(table, native).column('v').cast(pa.int64()).to_pylist()
== expected + [expected[0]]
+ assert _read(table, native, pb.equal('v', first),
['id']).column('id').to_pylist() == [0, len(values)]
+ assert _read(table, native, pb.is_null('v'),
['id']).column('id').to_pylist() == [len(values) - 1]
+
+
[email protected]('source_type,target_name,values,expected',
_TEMPORAL_CASES)
+def test_single_value_epoch_stats_do_not_prune_converted_match(
+ rest_catalog, source_type, target_name, values, expected):
+ catalog, _ = rest_catalog
+ _table(catalog, 'append', source_type, values[:1])
+ target = AtomicType(target_name)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
target)], False)
+ table = catalog.get_table('default.t')
+ literal = pa.scalar(expected[0],
type=PyarrowFieldParser.from_paimon_type(target)).as_py()
+ predicate = table.new_read_builder().new_predicate_builder().equal('v',
literal)
+ for native in (False, True):
+ assert _read(table, native, predicate,
['id']).column('id').to_pylist() == [0]
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('integer_type', [pa.int32(), pa.int64()])
+def test_nested_epoch_seconds_evolution_and_projection(rest_catalog, mode,
integer_type):
+ catalog, _ = rest_catalog
+ schema = pa.schema([
+ ('id', pa.int64()),
+ ('r', pa.struct([('t', pa.timestamp('us')), ('s', integer_type)])),
+ ('a', pa.list_(pa.timestamp('ns'))),
+ ('m', pa.map_(pa.string(), integer_type)),
+ ])
+ options = {'file.format': 'parquet', 'read.batch-size': '1'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif mode == 'de':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] if mode == 'pk' else [], options=options),
False)
+ boundary = 2 ** 63 - 1 if integer_type == pa.int64() else -(2 ** 31)
+ expected_boundary = -1000000 if integer_type == pa.int64() else
-2147483648000000
+ _write(catalog.get_table('default.t'), pa.Table.from_pylist([
+ {'id': 0, 'r': {'t': 42000001, 's': 42}, 'a': [42000000001, None, -1],
+ 'm': [('positive', 42), ('negative', -42)]},
+ {'id': 1, 'r': None, 'a': None, 'm': None},
+ {'id': 2, 'r': {'t': -1, 's': boundary}, 'a': [], 'm': []},
+ ], schema=schema))
+ catalog.alter_table('default.t', [
+ SchemaChange.update_column_type(['r', 't'], AtomicType('BIGINT')),
+ SchemaChange.update_column_type(['r', 's'],
AtomicType('TIMESTAMP(6)')),
+ SchemaChange.update_column_type(['a', 'element'],
AtomicType('BIGINT')),
+ SchemaChange.update_column_type(['m', 'value'],
AtomicType('TIMESTAMP(3)')),
+ ], False)
+ table = catalog.get_table('default.t')
+ expected_map = pa.array([[('positive', 42000), ('negative', -42000)],
None, []],
+ type=pa.map_(pa.string(),
pa.timestamp('ms'))).to_pylist()
+ for native in (False, True):
+ actual = _read(table, native)
+ assert actual.column('r').flatten()[0].to_pylist() == [42, None, -1]
+ assert actual.column('r').flatten()[1].cast(pa.int64()).to_pylist() ==
[42000000, None, expected_boundary]
+ assert actual.column('a').to_pylist() == [[42, None, -1], None, []]
+ assert actual.column('m').to_pylist() == expected_map
+ leaves = _read(table, native, projection=['id', 'r.t', 'r.s'])
+ assert leaves.column('r_t').to_pylist() == [42, None, -1]
+ assert leaves.column('r_s').cast(pa.int64()).to_pylist() == [42000000,
None, expected_boundary]
+ selected = _read(table, native, projection=[
+ 'id', "m['positive']", "m['negative']", "m['missing']"])
+ assert selected.column('m_positive').cast(pa.int64()).to_pylist() ==
[42000, None, None]
+ assert selected.column('m_negative').cast(pa.int64()).to_pylist() ==
[-42000, None, None]
+ assert selected.column('m_missing').null_count == 3
diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
index 1b6e4f817b..97d3ecc3a4 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
@@ -54,3 +54,61 @@ def test_timestamp_floor_overflow_returns_null(zone,
precision):
source, source.type, AtomicType(name + '(9)'), AtomicType('%s(%d)' %
(name, precision)))
quantum = 10 ** (9 - precision)
assert output.cast(pa.int64()).to_pylist() == [None, (-(2 ** 63) + 1000)
// quantum * quantum, -quantum, None]
+
+
[email protected]('unit,ticks', [('s', 1), ('ms', 1000), ('us',
1000000), ('ns', 1000000000)])
[email protected]('zone', [None, 'UTC'])
[email protected]('chunked', [False, True])
+def test_timestamp_to_bigint_uses_epoch_seconds_flooring(unit, ticks, zone,
chunked):
+ fraction = 0 if unit == 's' else 1
+ values = [ticks + fraction, -1, -ticks - fraction, 0,
+ (2 ** 31 + 2) * ticks, -(2 ** 31 + 2) * ticks, None]
+ expected = [1, -1, -1 if unit == 's' else -2, 0, 2 ** 31 + 2, -(2 ** 31 +
2), None]
+ source = pa.array([None] + values + [None], type=pa.timestamp(unit,
zone)).slice(1, len(values))
+ if chunked:
+ source = pa.chunked_array([source.slice(0, 2), source.slice(2, 0),
source.slice(2)])
+ result = cast_array_for_schema_evolution(source, pa.int64())
+ assert result.type == pa.int64()
+ assert result.to_pylist() == expected
+
+
[email protected]('unit,ticks', [('s', 1), ('ms', 1000), ('us',
1000000), ('ns', 1000000000)])
[email protected]('chunked', [False, True])
[email protected]('source_type', [pa.int32(), pa.int64()])
+def
test_integer_to_timestamp_uses_epoch_seconds_and_java_millis_wrapping(unit,
ticks, chunked, source_type):
+ values = [42, -42, 1700000000, None]
+ expected = [42 * ticks, -42 * ticks, 1700000000 * ticks, None]
+ if source_type == pa.int64():
+ values.extend([2 ** 63 - 1, -(2 ** 63), 2 ** 63 - 2])
+ expected.extend([-ticks, 0, -2 * ticks])
+ else:
+ values.extend([2 ** 31 - 1, -(2 ** 31), 0])
+ expected.extend([(2 ** 31 - 1) * ticks, -(2 ** 31) * ticks, 0])
+ source = pa.array([None] + values + [None], type=source_type).slice(1,
len(values))
+ if chunked:
+ source = pa.chunked_array([source.slice(0, 1), source.slice(1)])
+ result = cast_array_for_schema_evolution(source, pa.timestamp(unit))
+ assert result.type == pa.timestamp(unit)
+ assert result.cast(pa.int64()).to_pylist() == expected
+
+
[email protected]('zone,expected', [
+ (None, [9223372036854775, -9223372036854776, None]),
+ ('UTC', [-53757322505225, 53757322505224, None]),
+])
+def test_timestamp_to_bigint_matches_java_extreme_millis_round_trip(zone,
expected):
+ # Golden outputs from Java CastExecutors, including LTZ's int epoch day.
+ source = pa.array([2 ** 63 - 1, -(2 ** 63), None], type=pa.timestamp('ms',
zone))
+ assert cast_array_for_schema_evolution(source, pa.int64()).to_pylist() ==
expected
+
+
[email protected]('unit,expected', [
+ ('s', [9223372036854775, -9223372036854775, None]),
+ ('ms', [9223372036854775000, -9223372036854775000, None]),
+ ('us', [None, None, None]),
+ ('ns', [None, None, None]),
+])
+def test_bigint_to_timestamp_keeps_lenient_arrow_storage_overflow(unit,
expected):
+ source = pa.array([9223372036854775, -9223372036854775, None],
type=pa.int64())
+ result = cast_array_for_schema_evolution(source, pa.timestamp(unit))
+ assert result.cast(pa.int64()).to_pylist() == expected
diff --git a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
index 74bfec80f3..899634fafe 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
@@ -764,6 +764,15 @@ class SupportsCastTest(unittest.TestCase):
self.assertFalse(supports_cast(AtomicType(src), AtomicType(dst)),
'{} -> {}'.format(src, dst))
+ def test_can_execute_integer_to_timestamp(self):
+ for source in ('INT', 'BIGINT'):
+ for precision in (0, 1, 4, 7):
+ target = 'TIMESTAMP({})'.format(precision)
+ with self.subTest(source=source, target=target):
+ self.assertTrue(can_execute_cast(AtomicType(source),
AtomicType(target)))
+ for source in ('TINYINT', 'SMALLINT'):
+ self.assertFalse(can_execute_cast(AtomicType(source),
AtomicType('TIMESTAMP(6)')))
+
def test_can_execute_cast_decimal_precision(self):
# A numeric -> decimal cast has a PyArrow kernel but is only executable
# when the target precision can hold the source's range at the target
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 2eb425e156..2187e824df 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -73,6 +73,36 @@ def
test_write_builder_creates_row_id_updater_with_commit_identity(
table.new_read_builder.assert_not_called()
[email protected]('missing_method,option', [
+ ('data_evolution_enabled', 'data-evolution.enabled'),
+ ('row_tracking_enabled', 'row-tracking.enabled'),
+ ('deletion_vectors_enabled', 'deletion-vectors.enabled'),
+])
+def
test_python_row_id_delete_reports_unsupported_table_capabilities(missing_method,
option):
+ from pypaimon.write.table_delete import TableDeleteByRowId
+
+ table = mock.Mock()
+ table.options.data_evolution_enabled.return_value = True
+ table.options.row_tracking_enabled.return_value = True
+ table.options.deletion_vectors_enabled.return_value = True
+ getattr(table.options, missing_method).return_value = False
+ with pytest.raises(NotImplementedError, match=option):
+ TableDeleteByRowId(table).delete([])
+ table.snapshot_manager.assert_not_called()
+
+
+def test_python_row_id_delete_keeps_invalid_input_as_value_error():
+ from pypaimon.write.table_delete import TableDeleteByRowId
+
+ table = mock.Mock()
+ table.options.data_evolution_enabled.return_value = True
+ table.options.row_tracking_enabled.return_value = True
+ table.options.deletion_vectors_enabled.return_value = True
+ with pytest.raises(ValueError, match='_ROW_ID value must not be null'):
+ TableDeleteByRowId(table).delete([None])
+ table.snapshot_manager.assert_not_called()
+
+
def test_batch_row_id_update_batches_reuse_file_index():
table = mock.MagicMock()
table.field_names = ["value"]
@@ -922,7 +952,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
table = self._create_seeded_table()
pb = table.new_read_builder().new_predicate_builder()
- with self.assertRaises(ValueError) as ctx:
+ with self.assertRaises(NotImplementedError) as ctx:
self._do_delete_by_predicate(table, pb.equal('id', 1))
self.assertIn('deletion-vectors.enabled', str(ctx.exception))
@@ -940,7 +970,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
def test_delete_by_row_id_requires_deletion_vectors(self):
table = self._create_seeded_table()
- with self.assertRaises(ValueError) as ctx:
+ with self.assertRaises(NotImplementedError) as ctx:
self._do_delete_by_row_id(table, [0])
self.assertIn('deletion-vectors.enabled', str(ctx.exception))
@@ -1066,7 +1096,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
pb.equal('age', 30),
])
- with self.assertRaises(ValueError) as ctx:
+ with self.assertRaises(NotImplementedError) as ctx:
self._do_delete_by_predicate(table, predicate)
self.assertIn('deletion-vectors.enabled', str(ctx.exception))
diff --git a/paimon-python/pypaimon/write/table_delete.py
b/paimon-python/pypaimon/write/table_delete.py
index 00ea6b1cec..2dd78d351b 100644
--- a/paimon-python/pypaimon/write/table_delete.py
+++ b/paimon-python/pypaimon/write/table_delete.py
@@ -98,16 +98,18 @@ class TableDeleteByRowId:
return commit_messages
def _validate_delete(self):
+ # Java DataEvolutionDeleteSink.validateTable reports unsupported table
+ # capabilities; Native maps the same failure to NotImplementedError.
if not self.table.options.data_evolution_enabled():
- raise ValueError(
+ raise NotImplementedError(
"delete requires 'data-evolution.enabled' = 'true'."
)
if not self.table.options.row_tracking_enabled():
- raise ValueError(
+ raise NotImplementedError(
"delete requires 'row-tracking.enabled' = 'true'."
)
if not self.table.options.deletion_vectors_enabled(False):
- raise ValueError(
+ raise NotImplementedError(
"delete requires 'deletion-vectors.enabled' = 'true'."
)
