[python] Add efficient VARIANT path get and replace

docs/docs/pypaimon/python-api.mdx

@@ -1126,6 +1126,28 @@ Supported Paimon type strings for shredded sub-fields: `BOOLEAN`, `INT`, `BIGINT
 
 </Tabs>
 
+### VARIANT Path Updates
+
+Read existing paths as Arrow arrays, use Arrow compute, and replace them
+without decoding unrelated fields:
+
+```python
+import pyarrow as pa
+import pyarrow.compute as pc
+
+from pypaimon.data import variant_get, variant_replace
+
+current = variant_get(payload, '$.velocity.y', pa.float64())
+updated_payload = variant_replace(
+    payload, '$.velocity.y', pc.negate(current))
+```
+
+The Arrow type must match the stored VARIANT type; these APIs do not cast
+values. Exact extraction supports scalar types and nested struct, list, and
+string-keyed map types. Replacement supports scalar types. Missing paths read
+as NULL and remain unchanged unless `strict=True` is specified. Pass mappings
+to process multiple paths in one pass.
+
 
 **`GenericVariant` API:**
 

paimon-python/pypaimon/data/__init__.py

@@ -17,5 +17,11 @@
 
 from pypaimon.data.timestamp import Timestamp
 from pypaimon.data.decimal import Decimal
+from pypaimon.data.variant_path import variant_get, variant_replace
 
-__all__ = ['Timestamp', 'Decimal']
+__all__ = [
+    'Timestamp',
+    'Decimal',
+    'variant_get',
+    'variant_replace',
+]

paimon-python/pypaimon/data/generic_variant.py

@@ -148,6 +148,12 @@ def _write_le(buf, pos, value, n):
     buf[pos:pos + n] = value.to_bytes(n, 'little')
 
 
+def _decimal_from_unscaled(unscaled, scale):
+    sign = 1 if unscaled < 0 else 0
+    digits = tuple(int(digit) for digit in str(abs(unscaled))) or (0,)
+    return _decimal.Decimal((sign, digits, -scale))
+
+
 def _short_str_header(size):
     return (size << 2) | _SHORT_STR
 
@@ -306,18 +312,26 @@ def append_float(self, f):
         self._pos += 4
 
     def append_decimal(self, d):
-        d = d.normalize()
         sign, digits, exponent = d.as_tuple()
-        if exponent > 0:
-            raise ValueError(
-                f'append_decimal requires a non-positive exponent (got {d!r}); '
-                'use append_double() for Decimal values with positive exponents'
-            )
         unscaled = int(''.join(str(x) for x in digits))
         if sign:
             unscaled = -unscaled
-        scale = -exponent if exponent < 0 else 0
-        precision = len(digits)
+        if exponent > 0:
+            unscaled *= 10 ** exponent
+            scale = 0
+        else:
+            scale = -exponent
+        self.append_decimal_unscaled(
+            unscaled, max(1, len(str(abs(unscaled)))), scale)
+
+    def append_decimal_unscaled(self, unscaled, precision, scale):
+        if not 0 <= scale <= _MAX_DECIMAL16_PRECISION:
+            raise ValueError(f'Unsupported VARIANT decimal scale: {scale}')
+        if not 0 < precision <= _MAX_DECIMAL16_PRECISION:
+            raise ValueError(
+                f'Unsupported VARIANT decimal precision: {precision}')
+        if not -(1 << 127) <= unscaled < (1 << 127):
+            raise ValueError('VARIANT decimal value exceeds 128 bits')
 
         if scale <= _MAX_DECIMAL4_PRECISION and precision <= _MAX_DECIMAL4_PRECISION:
             self._write_byte(_primitive_header(_DECIMAL4))
@@ -668,7 +682,7 @@ def _to_python_impl(self, value, metadata, pos):
             else:
                 raw = bytes(value[pos + 2:pos + 18])
                 unscaled = int.from_bytes(raw, 'little', signed=True)
-            return _decimal.Decimal(unscaled) / (_decimal.Decimal(10) ** scale)
+            return _decimal_from_unscaled(unscaled, scale)
         if vtype == _Type.STRING:
             if basic_type == _SHORT_STR:
                 return value[pos + 1:pos + 1 + type_info].decode('utf-8')