[python] Support UpdateColumnDefaultValue in SchemaManager

Apply UpdateColumnDefaultValue when committing schema changes, so a column
default can be set, changed and cleared through the Python SchemaManager.

Validate the value against the target type before storing it, mirroring the
Java DefaultValueUtils string casts for booleans, numbers, dates, times and
timestamps. Widen new_default_value to Optional[str], matching the @nullable
contract on the Java side, where None clears the default.

ARRAY, MAP and ROW defaults are accepted by Java but are not parsed here yet,
so they are rejected with an explicit message.

Author: wenchao.wu

paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java

@@ -37,6 +37,7 @@
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.data.variant.GenericVariant;
 import org.apache.paimon.data.variant.GenericVariantBuilder;
@@ -85,6 +86,7 @@
 import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.RowKind;
@@ -107,6 +109,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -732,6 +735,70 @@ public void testReadPythonDynamicBucketHashIndex() throws Exception {
                         "hello-java, 42, java-new", "python-only, 7, python-only");
     }
 
+    @Test
+    @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    public void testReadPythonColumnDefaultValues() throws Exception {
+        FileStoreTable table =
+                (FileStoreTable) catalog.getTable(identifier("column_defaults_python_to_java"));
+        assertThat(table.rowType().getFields())
+                .extracting(DataField::defaultValue)
+                .containsExactly(
+                        null,
+                        "12:30+01",
+                        "2024-02-31 24:00:00",
+                        "+12345-01-01",
+                        "2024-01-01T24:00",
+                        "0e-2147483645",
+                        "12345678901",
+                        "1700000000000",
+                        "''");
+
+        // Opening the writer casts every default; null fields take their default.
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            write.write(GenericRow.of(1, null, null, null, null, null, null, null, null));
+            commit.commit(write.prepareCommit());
+        }
+
+        List<String> result =
+                getResult(
+                        table.newRead(),
+                        table.newScan().plan().splits(),
+                        row ->
+                                row.getInt(0)
+                                        + ", "
+                                        + row.getInt(1)
+                                        + ", "
+                                        + row.getTimestamp(2, 3)
+                                        + ", "
+                                        + row.getTimestamp(3, 3)
+                                        + ", "
+                                        + row.getTimestamp(4, 3)
+                                        + ", "
+                                        + row.getDecimal(5, 10, 2)
+                                        + ", "
+                                        + row.isNullAt(6)
+                                        + ", "
+                                        + row.getTimestamp(7, 3)
+                                        + ", ["
+                                        + row.getString(8)
+                                        + "]");
+        Timestamp nextLocalMidnight =
+                Timestamp.fromInstant(
+                        LocalDateTime.of(2024, 1, 2, 0, 0)
+                                .atZone(ZoneId.systemDefault())
+                                .toInstant());
+        // parseTime adds a '+hh' offset to the hour, so 12:30+01 is 13:30.
+        assertThat(result)
+                .containsExactly(
+                        "1, "
+                                + (13 * 3600 + 30 * 60) * 1000
+                                + ", 2024-03-01T00:00, +12345-01-01T00:00, "
+                                + nextLocalMidnight
+                                + ", 0.00, true, 2023-11-14T22:13:20, []");
+    }
+
     private int assignDynamicBucket(FileStoreTable table, InternalRow row) {
         InternalRowSerializer serializer =
                 new InternalRowSerializer(DataTypes.STRING(), DataTypes.BIGINT());

paimon-python/dev/run_mixed_tests.sh

@@ -248,7 +248,7 @@ run_python_write_test() {
 
     # Run the parameterized Python test method for writing data (pk table, includes bucket num assertion)
     echo "Running Python test for JavaPyReadWriteTest (test_py_write_read_pk_table)..."
-    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_read_append_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence" -v; then
+    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k "test_py_write_read_pk_table or test_py_write_read_append_table or test_py_write_dynamic_bucket_hash_index or test_py_write_floating_sequence or test_py_write_column_default_values" -v; then
         echo -e "${GREEN}✓ Python write test completed successfully${NC}"
         return 0
     else
@@ -267,7 +267,7 @@ run_java_read_test() {
     echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read Parquet/Orc/Avro)..."
     echo "Note: Maven may download dependencies on first run, this may take a while..."
     local parquet_result=0
-    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadAppendTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
+    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadAppendTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence+testReadPythonColumnDefaultValues -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
         echo -e "${GREEN}✓ Java read Parquet/Orc/Avro test completed successfully${NC}"
     else
         echo -e "${RED}✗ Java read Parquet/Orc/Avro test failed${NC}"