fix

Author: yuzelin

paimon_python_java/paimon-python-java-bridge/src/main/java/org/apache/paimon/python/BytesWriter.java

@@ -32,7 +32,6 @@
 
 import java.io.ByteArrayInputStream;
 import java.util.List;
-import java.util.Objects;
 import java.util.stream.Collectors;
 
 /** Write Arrow bytes to Paimon. */
@@ -53,11 +52,11 @@ public BytesWriter(TableWrite tableWrite, RowType rowType) {
                         .collect(Collectors.toList());
     }
 
-    public void write(byte[] bytes, boolean needCheckSchema) throws Exception {
+    public void write(byte[] bytes) throws Exception {
         ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
         ArrowStreamReader arrowStreamReader = new ArrowStreamReader(bais, allocator);
         VectorSchemaRoot vsr = arrowStreamReader.getVectorSchemaRoot();
-        if (needCheckSchema && !checkSchema(arrowFields, vsr.getSchema().getFields())) {
+        if (!checkTypesIgnoreNullability(arrowFields, vsr.getSchema().getFields())) {
             throw new RuntimeException(
                     String.format(
                             "Input schema isn't consistent with table schema.\n"
@@ -79,25 +78,22 @@ public void close() {
         allocator.close();
     }
 
-    private boolean checkSchema(List<Field> expectedFields, List<Field> actualFields) {
+    private boolean checkTypesIgnoreNullability(
+            List<Field> expectedFields, List<Field> actualFields) {
         if (expectedFields.size() != actualFields.size()) {
             return false;
         }
 
         for (int i = 0; i < expectedFields.size(); i++) {
             Field expectedField = expectedFields.get(i);
             Field actualField = actualFields.get(i);
-            if (!checkFieldIgnoreNullability(expectedField, actualField)
+            // ArrowType doesn't have nullability (similar to DataTypeRoot)
+            if (!actualField.getType().equals(expectedField.getType())
                     || !checkSchema(expectedField.getChildren(), actualField.getChildren())) {
                 return false;
             }
         }
 
         return true;
     }
-
-    private boolean checkFieldIgnoreNullability(Field expected, Field actual) {
-        return Objects.equals(expected.getName(), actual.getName())
-                && Objects.equals(expected.getType(), actual.getType());
-    }
 }

paimon_python_java/pypaimon.py

@@ -219,21 +219,21 @@ def __init__(self, j_batch_table_write, j_row_type, arrow_schema: pa.Schema):
     def write_arrow(self, table):
         for record_batch in table.to_reader():
             # TODO: can we use a reusable stream in #_write_arrow_batch ?
-            self._write_arrow_batch(record_batch, True)
+            self._write_arrow_batch(record_batch)
 
     def write_arrow_batch(self, record_batch):
-        self._write_arrow_batch(record_batch, True)
+        self._write_arrow_batch(record_batch)
 
     def write_pandas(self, dataframe: pd.DataFrame):
         record_batch = pa.RecordBatch.from_pandas(dataframe, schema=self._arrow_schema)
-        self._write_arrow_batch(record_batch, False)
+        self._write_arrow_batch(record_batch)
 
-    def _write_arrow_batch(self, record_batch, check_schema):
+    def _write_arrow_batch(self, record_batch):
         stream = pa.BufferOutputStream()
         with pa.RecordBatchStreamWriter(stream, record_batch.schema) as writer:
             writer.write(record_batch)
         arrow_bytes = stream.getvalue().to_pybytes()
-        self._j_bytes_writer.write(arrow_bytes, check_schema)
+        self._j_bytes_writer.write(arrow_bytes)
 
     def prepare_commit(self) -> List['CommitMessage']:
         j_commit_messages = self._j_batch_table_write.prepareCommit()