From 0a78244d956aa5b2c54d3cc0d13948f93cf85c25 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 15:30:00 +0800
Subject: [PATCH 01/17] fix fastavro read java avro issue

---
 .../manifest/fastavro_zstd_py36_fix.py        | 59 +++++++++++++++++++
 .../pypaimon/manifest/index_manifest_file.py  |  7 +++
 .../manifest/manifest_file_manager.py         |  7 +++
 .../manifest/manifest_list_manager.py         |  8 +++
 4 files changed, 81 insertions(+)
 create mode 100644 paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py

diff --git a/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py b/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py
new file mode 100644
index 000000000000..39af7cbd371d
--- /dev/null
+++ b/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py
@@ -0,0 +1,59 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+################################################################################
+
+
+import sys
+
+if sys.version_info[:2] == (3, 6):
+    try:
+        import fastavro._read as fastavro_read
+        import zstandard as zstd
+        from io import BytesIO
+        
+        _original_zstandard_read_block = None
+        if hasattr(fastavro_read, 'zstandard_read_block'):
+            _original_zstandard_read_block = fastavro_read.zstandard_read_block
+        
+        def _fixed_zstandard_read_block(decoder):
+            from fastavro._read import read_long
+            length = read_long(decoder)
+            data = decoder.read_fixed(length)
+            
+            try:
+                decompressor = zstd.ZstdDecompressor()
+                
+                try:
+                    decompressed = decompressor.decompress(data)
+                    return BytesIO(decompressed)
+                except zstd.ZstdError as e:
+                    error_msg = str(e).lower()
+                    if "could not determine content size" in error_msg or "content size" in error_msg:
+                        dctx = zstd.ZstdDecompressor()
+                        with dctx.stream_reader(BytesIO(data)) as reader:
+                            decompressed = reader.read()
+                            return BytesIO(decompressed)
+                    else:
+                        raise
+            except (zstd.ZstdError, Exception) as e:
+                raise zstd.ZstdError(f"Failed to decompress zstd data on Python 3.6: {e}") from e
+        
+        if hasattr(fastavro_read, 'zstandard_read_block'):
+            fastavro_read.zstandard_read_block = _fixed_zstandard_read_block
+    except (ImportError, AttributeError):
+        pass
+
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py
index cd66ee5fce7e..a326d5dedabf 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -17,6 +17,13 @@
 
 from io import BytesIO
 from typing import List
+import sys
+
+# Apply zstd fix for Python 3.6 before importing fastavro
+try:
+    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+except ImportError:
+    pass
 
 import fastavro
 
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index f6ae41e3d386..eaa2c927fd54 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -18,6 +18,13 @@
 from concurrent.futures import ThreadPoolExecutor
 from io import BytesIO
 from typing import List
+import sys
+
+# Apply zstd fix for Python 3.6 before importing fastavro
+try:
+    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+except ImportError:
+    pass
 
 import fastavro
 
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index a1897fbee721..017b18d6cd43 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -18,6 +18,13 @@
 
 from io import BytesIO
 from typing import List
+import sys
+
+# Apply zstd fix for Python 3.6 before importing fastavro
+try:
+    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+except ImportError:
+    pass
 
 import fastavro
 
@@ -59,6 +66,7 @@ def read(self, manifest_list_name: str) -> List[ManifestFileMeta]:
             avro_bytes = input_stream.read()
         buffer = BytesIO(avro_bytes)
         reader = fastavro.reader(buffer)
+        
         for record in reader:
             stats_dict = dict(record['_PARTITION_STATS'])
             partition_stats = SimpleStats(

From ec958f775d158ad322befe29c99a39427263dec3 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 18:16:24 +0800
Subject: [PATCH 02/17] fastavro compatible with zstd for py36 hack

---
 .../java/org/apache/paimon/JavaPyE2ETest.java | 43 ++++++++++++++
 paimon-python/dev/lint-python.sh              |  4 --
 paimon-python/dev/run_mixed_tests.sh          | 59 +++++++++++++++++--
 .../manifest/fastavro_zstd_py36_fix.py        | 59 -------------------
 .../pypaimon/manifest/index_manifest_file.py  |  9 ++-
 .../manifest/manifest_file_manager.py         |  8 ++-
 .../manifest/manifest_list_manager.py         |  9 ++-
 .../read/reader/format_avro_reader.py         | 10 ++++
 .../tests/e2e/java_py_read_write_test.py      | 22 +++++++
 9 files changed, 146 insertions(+), 77 deletions(-)
 delete mode 100644 paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py

diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 39c2011bf486..809ff7bb81e9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -67,6 +67,7 @@
 
 import static org.apache.paimon.CoreOptions.BUCKET;
 import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
+import static org.apache.paimon.CoreOptions.MANIFEST_COMPRESSION;
 import static org.apache.paimon.CoreOptions.TARGET_FILE_SIZE;
 import static org.apache.paimon.data.DataFormatTestUtil.internalRowToString;
 import static org.apache.paimon.table.SimpleTableTestBase.getResult;
@@ -379,6 +380,48 @@ public void testReadPkTable() throws Exception {
                         "6, Beef, Meat, 8.0");
     }
 
+    @Test
+    @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    public void testJavaWriteZstdManifestTable() throws Exception {
+        Identifier identifier = identifier("zstd_manifest_test_table");
+
+        Schema schema =
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT())
+                        .column("name", DataTypes.STRING())
+                        .column("value", DataTypes.DOUBLE())
+                        .option(
+                                MANIFEST_COMPRESSION.key(),
+                                "zstd")
+                        .build();
+
+        catalog.createTable(identifier, schema, true);
+        Table table = catalog.getTable(identifier);
+        FileStoreTable fileStoreTable = (FileStoreTable) table;
+
+        String manifestCompression = fileStoreTable.coreOptions().manifestCompression();
+        assertThat(manifestCompression).isEqualTo("zstd");
+
+        try (StreamTableWrite write = fileStoreTable.newWrite(commitUser);
+                InnerTableCommit commit = fileStoreTable.newCommit(commitUser)) {
+
+            write.write(GenericRow.of(1, BinaryString.fromString("test1"), 1.1));
+            write.write(GenericRow.of(2, BinaryString.fromString("test2"), 2.2));
+            write.write(GenericRow.of(3, BinaryString.fromString("test3"), 3.3));
+
+            commit.commit(0, write.prepareCommit(true, 0));
+        }
+
+        // Verify data was written
+        assertThat(fileStoreTable.newSnapshotReader().read().dataSplits()).hasSizeGreaterThan(0);
+
+        System.out.println(
+                "✓ Java test completed: Created table with zstd-compressed manifest files");
+        System.out.println("  Table location: " + fileStoreTable.location());
+        System.out.println("  Manifest compression: " + manifestCompression);
+        System.out.println("  Python test should now read this table to reproduce the error");
+    }
+
     // Helper method from TableTestBase
     protected Identifier identifier(String tableName) {
         return new Identifier(database, tableName);
diff --git a/paimon-python/dev/lint-python.sh b/paimon-python/dev/lint-python.sh
index 469ee56c9d95..d174b120ad4f 100755
--- a/paimon-python/dev/lint-python.sh
+++ b/paimon-python/dev/lint-python.sh
@@ -203,10 +203,6 @@ function mixed_check() {
     # Get Python version
     PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')")
     echo "Detected Python version: $PYTHON_VERSION"
-    if [ "$PYTHON_VERSION" = "3.6" ]; then
-        print_function "STAGE" "mixed tests checks... [SKIPPED]"
-        return
-    fi
     print_function "STAGE" "mixed tests checks"
 
     # Path to the mixed tests script
diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index 337b694e0c7c..2483162f32a6 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -192,6 +192,40 @@ run_pk_dv_test() {
         return 1
     fi
 }
+
+# Function to run zstd manifest e2e test
+run_zstd_manifest_test() {
+    echo -e "${YELLOW}=== Step 6: Running Zstd Manifest Compression Test (Python 3.6 Compatibility) ===${NC}"
+
+    cd "$PROJECT_ROOT"
+
+    echo "Running Maven test for JavaPyE2ETest.testJavaWriteZstdManifestTable..."
+    local java_result=0
+    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testJavaWriteZstdManifestTable -pl paimon-core -Drun.e2e.tests=true; then
+        echo -e "${GREEN}✓ Java zstd manifest test completed successfully${NC}"
+    else
+        echo -e "${RED}✗ Java zstd manifest test failed${NC}"
+        java_result=1
+    fi
+
+    echo ""
+
+    cd "$PAIMON_PYTHON_ROOT"
+    echo "Running Python test for JavaPyReadWriteTest.test_read_zstd_manifest_table..."
+    local python_result=0
+    if python -m pytest pypaimon/tests/e2e/java_py_read_write_test.py::JavaPyReadWriteTest::test_read_zstd_manifest_table -v; then
+        echo -e "${GREEN}✓ Python zstd manifest test completed successfully${NC}"
+    else
+        echo -e "${RED}✗ Python zstd manifest test failed${NC}"
+        python_result=1
+    fi
+
+    if [[ $java_result -eq 0 && $python_result -eq 0 ]]; then
+        return 0
+    else
+        return 1
+    fi
+}
 # Main execution
 main() {
     local java_write_result=0
@@ -199,6 +233,7 @@ main() {
     local python_write_result=0
     local java_read_result=0
     local pk_dv_result=0
+    local zstd_manifest_result=0
 
     echo -e "${YELLOW}Starting mixed language test execution...${NC}"
     echo ""
@@ -241,6 +276,12 @@ main() {
     if ! run_pk_dv_test; then
         pk_dv_result=1
     fi
+
+    # Run zstd manifest test
+    if ! run_zstd_manifest_test; then
+        zstd_manifest_result=1
+    fi
+
     echo ""
     echo -e "${YELLOW}=== Test Results Summary ===${NC}"
 
@@ -268,18 +309,24 @@ main() {
         echo -e "${RED}✗ Java Read Test (Parquet + Lance): FAILED${NC}"
     fi
 
-     if [[ $pk_dv_result -eq 0 ]]; then
-          echo -e "${GREEN}✓ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): PASSED${NC}"
-      else
-          echo -e "${RED}✗ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): FAILED${NC}"
-      fi
+    if [[ $pk_dv_result -eq 0 ]]; then
+        echo -e "${GREEN}✓ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): PASSED${NC}"
+    else
+        echo -e "${RED}✗ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): FAILED${NC}"
+    fi
+
+    if [[ $zstd_manifest_result -eq 0 ]]; then
+        echo -e "${GREEN}✓ Zstd Manifest Test (Python 3.6 Compatibility): PASSED${NC}"
+    else
+        echo -e "${RED}✗ Zstd Manifest Test (Python 3.6 Compatibility): FAILED${NC}"
+    fi
 
     echo ""
 
     # Clean up warehouse directory after all tests
     cleanup_warehouse
 
-    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 ]]; then
+    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 && $zstd_manifest_result -eq 0 ]]; then
         echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability verified.${NC}"
         return 0
     else
diff --git a/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py b/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py
deleted file mode 100644
index 39af7cbd371d..000000000000
--- a/paimon-python/pypaimon/manifest/fastavro_zstd_py36_fix.py
+++ /dev/null
@@ -1,59 +0,0 @@
-################################################################################
-#  Licensed to the Apache Software Foundation (ASF) under one
-#  or more contributor license agreements.  See the NOTICE file
-#  distributed with this work for additional information
-#  regarding copyright ownership.  The ASF licenses this file
-#  to you under the Apache License, Version 2.0 (the
-#  "License"); you may not use this file except in compliance
-#  with the License.  You may obtain a copy of the License at
-#
-#      http://www.apache.org/licenses/LICENSE-2.0
-#
-#  Unless required by applicable law or agreed to in writing, software
-#  distributed under the License is distributed on an "AS IS" BASIS,
-#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-#  See the License for the specific language governing permissions and
-#  limitations under the License.
-################################################################################
-
-
-import sys
-
-if sys.version_info[:2] == (3, 6):
-    try:
-        import fastavro._read as fastavro_read
-        import zstandard as zstd
-        from io import BytesIO
-        
-        _original_zstandard_read_block = None
-        if hasattr(fastavro_read, 'zstandard_read_block'):
-            _original_zstandard_read_block = fastavro_read.zstandard_read_block
-        
-        def _fixed_zstandard_read_block(decoder):
-            from fastavro._read import read_long
-            length = read_long(decoder)
-            data = decoder.read_fixed(length)
-            
-            try:
-                decompressor = zstd.ZstdDecompressor()
-                
-                try:
-                    decompressed = decompressor.decompress(data)
-                    return BytesIO(decompressed)
-                except zstd.ZstdError as e:
-                    error_msg = str(e).lower()
-                    if "could not determine content size" in error_msg or "content size" in error_msg:
-                        dctx = zstd.ZstdDecompressor()
-                        with dctx.stream_reader(BytesIO(data)) as reader:
-                            decompressed = reader.read()
-                            return BytesIO(decompressed)
-                    else:
-                        raise
-            except (zstd.ZstdError, Exception) as e:
-                raise zstd.ZstdError(f"Failed to decompress zstd data on Python 3.6: {e}") from e
-        
-        if hasattr(fastavro_read, 'zstandard_read_block'):
-            fastavro_read.zstandard_read_block = _fixed_zstandard_read_block
-    except (ImportError, AttributeError):
-        pass
-
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py
index a326d5dedabf..cd764eacc857 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -17,16 +17,19 @@
 
 from io import BytesIO
 from typing import List
-import sys
 
-# Apply zstd fix for Python 3.6 before importing fastavro
 try:
-    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+    from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
     pass
 
 import fastavro
 
+try:
+    fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError):
+    pass
+
 from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
 from pypaimon.index.index_file_meta import IndexFileMeta
 from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index eaa2c927fd54..b78756c94ffb 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -20,14 +20,18 @@
 from typing import List
 import sys
 
-# Apply zstd fix for Python 3.6 before importing fastavro
 try:
-    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+    from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
     pass
 
 import fastavro
 
+try:
+    fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError):
+    pass
+
 from datetime import datetime
 
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index 017b18d6cd43..c7533d018e5a 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -18,16 +18,19 @@
 
 from io import BytesIO
 from typing import List
-import sys
 
-# Apply zstd fix for Python 3.6 before importing fastavro
 try:
-    from pypaimon.manifest.fastavro_zstd_py36_fix import *  # noqa: F401, F403
+    from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
     pass
 
 import fastavro
 
+try:
+    fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError):
+    pass
+
 from pypaimon.manifest.schema.manifest_file_meta import (
     MANIFEST_FILE_META_SCHEMA, ManifestFileMeta)
 from pypaimon.manifest.schema.simple_stats import SimpleStats
diff --git a/paimon-python/pypaimon/read/reader/format_avro_reader.py b/paimon-python/pypaimon/read/reader/format_avro_reader.py
index 4114d8e93b59..be2806b63184 100644
--- a/paimon-python/pypaimon/read/reader/format_avro_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_avro_reader.py
@@ -18,7 +18,17 @@
 
 from typing import List, Optional, Any
 
+try:
+    from pypaimon.manifest import fastavro_py36_compat
+except ImportError:
+    pass
+
 import fastavro
+
+try:
+    fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError):
+    pass
 import pyarrow as pa
 import pyarrow.dataset as ds
 from pyarrow import RecordBatch
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index b88224784c4a..4cd8158d5162 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -249,3 +249,25 @@ def test_pk_dv_read_multi_batch_raw_convertable(self):
             'b': [i * 100 for i in range(1, 10001) if i * 10 != 81930]
         }, schema=pa_schema)
         self.assertEqual(expected, actual)
+
+    def test_read_zstd_manifest_table(self):
+        table_name = 'default.zstd_manifest_test_table'
+        
+        try:
+            table = self.catalog.get_table(table_name)
+        except Exception as e:
+            self.fail(f"Failed to get table {table_name}. "
+                     f"Make sure Java test (JavaPyZstdManifestTest.testJavaWriteZstdManifestTable) "
+                     f"has been run first. Error: {e}")
+
+        read_builder = table.new_read_builder()
+        table_scan = read_builder.new_scan()
+        table_read = read_builder.new_read()
+        
+        splits = table_scan.plan().splits()
+        result = table_read.to_pandas(splits)
+        
+        self.assertEqual(len(result), 3)
+        expected_ids = {1, 2, 3}
+        actual_ids = set(result['id'].tolist())
+        self.assertEqual(actual_ids, expected_ids)

From d25855bc48fd4529b913ece9cfc73d91fdc583f4 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 18:20:40 +0800
Subject: [PATCH 03/17] fix code format

---
 .../test/java/org/apache/paimon/JavaPyE2ETest.java    | 11 +----------
 1 file changed, 1 insertion(+), 10 deletions(-)

diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 809ff7bb81e9..907b263a7138 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -390,9 +390,7 @@ public void testJavaWriteZstdManifestTable() throws Exception {
                         .column("id", DataTypes.INT())
                         .column("name", DataTypes.STRING())
                         .column("value", DataTypes.DOUBLE())
-                        .option(
-                                MANIFEST_COMPRESSION.key(),
-                                "zstd")
+                        .option(MANIFEST_COMPRESSION.key(), "zstd")
                         .build();
 
         catalog.createTable(identifier, schema, true);
@@ -412,14 +410,7 @@ public void testJavaWriteZstdManifestTable() throws Exception {
             commit.commit(0, write.prepareCommit(true, 0));
         }
 
-        // Verify data was written
         assertThat(fileStoreTable.newSnapshotReader().read().dataSplits()).hasSizeGreaterThan(0);
-
-        System.out.println(
-                "✓ Java test completed: Created table with zstd-compressed manifest files");
-        System.out.println("  Table location: " + fileStoreTable.location());
-        System.out.println("  Manifest compression: " + manifestCompression);
-        System.out.println("  Python test should now read this table to reproduce the error");
     }
 
     // Helper method from TableTestBase

From 401ff26af6209d26f15d380c839c60b1f08132af Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 18:32:22 +0800
Subject: [PATCH 04/17] fix

---
 paimon-python/dev/run_mixed_tests.sh                     | 4 ++--
 paimon-python/pypaimon/manifest/index_manifest_file.py   | 7 ++++---
 paimon-python/pypaimon/manifest/manifest_file_manager.py | 7 ++++---
 paimon-python/pypaimon/manifest/manifest_list_manager.py | 7 ++++---
 paimon-python/pypaimon/read/reader/format_avro_reader.py | 7 ++++---
 5 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index 2483162f32a6..049d791fdc35 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -210,10 +210,10 @@ run_zstd_manifest_test() {
 
     echo ""
 
-    cd "$PAIMON_PYTHON_ROOT"
+    cd "$PAIMON_PYTHON_DIR"
     echo "Running Python test for JavaPyReadWriteTest.test_read_zstd_manifest_table..."
     local python_result=0
-    if python -m pytest pypaimon/tests/e2e/java_py_read_write_test.py::JavaPyReadWriteTest::test_read_zstd_manifest_table -v; then
+    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_zstd_manifest_table -v; then
         echo -e "${GREEN}✓ Python zstd manifest test completed successfully${NC}"
     else
         echo -e "${RED}✗ Python zstd manifest test failed${NC}"
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py
index cd764eacc857..8c1770e207ed 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -21,13 +21,14 @@
 try:
     from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
-    pass
+    fastavro_py36_compat = None
 
 import fastavro
 
 try:
-    fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError):
+    if fastavro_py36_compat is not None:
+        fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError, NameError):
     pass
 
 from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index b78756c94ffb..6fdb026345e5 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -23,13 +23,14 @@
 try:
     from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
-    pass
+    fastavro_py36_compat = None
 
 import fastavro
 
 try:
-    fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError):
+    if fastavro_py36_compat is not None:
+        fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError, NameError):
     pass
 
 from datetime import datetime
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index c7533d018e5a..99906a147607 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -22,13 +22,14 @@
 try:
     from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
-    pass
+    fastavro_py36_compat = None
 
 import fastavro
 
 try:
-    fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError):
+    if fastavro_py36_compat is not None:
+        fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError, NameError):
     pass
 
 from pypaimon.manifest.schema.manifest_file_meta import (
diff --git a/paimon-python/pypaimon/read/reader/format_avro_reader.py b/paimon-python/pypaimon/read/reader/format_avro_reader.py
index be2806b63184..32a9e283769e 100644
--- a/paimon-python/pypaimon/read/reader/format_avro_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_avro_reader.py
@@ -21,13 +21,14 @@
 try:
     from pypaimon.manifest import fastavro_py36_compat
 except ImportError:
-    pass
+    fastavro_py36_compat = None
 
 import fastavro
 
 try:
-    fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError):
+    if fastavro_py36_compat is not None:
+        fastavro_py36_compat._apply_zstd_patch()
+except (ImportError, AttributeError, NameError):
     pass
 import pyarrow as pa
 import pyarrow.dataset as ds

From 2d35e761a2b1d4f974ad04d2a93fa071a4ce1e6d Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 19:19:11 +0800
Subject: [PATCH 05/17] fix code format

---
 paimon-python/pypaimon/manifest/manifest_file_manager.py  | 1 -
 .../pypaimon/tests/e2e/java_py_read_write_test.py         | 8 +++++---
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 6fdb026345e5..c7269146bb6c 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -18,7 +18,6 @@
 from concurrent.futures import ThreadPoolExecutor
 from io import BytesIO
 from typing import List
-import sys
 
 try:
     from pypaimon.manifest import fastavro_py36_compat
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 4cd8158d5162..6a24dd1a4ce7 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -256,9 +256,11 @@ def test_read_zstd_manifest_table(self):
         try:
             table = self.catalog.get_table(table_name)
         except Exception as e:
-            self.fail(f"Failed to get table {table_name}. "
-                     f"Make sure Java test (JavaPyZstdManifestTest.testJavaWriteZstdManifestTable) "
-                     f"has been run first. Error: {e}")
+            self.fail(
+                f"Failed to get table {table_name}. "
+                f"Make sure Java test (JavaPyE2ETest.testJavaWriteZstdManifestTable) "
+                f"has been run first. Error: {e}"
+            )
 
         read_builder = table.new_read_builder()
         table_scan = read_builder.new_scan()

From 002acfb4b45380bf2006a08724bce8c3f538f547 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 20:25:19 +0800
Subject: [PATCH 06/17] fix CI for py36

---
 paimon-python/dev/run_mixed_tests.sh          | 29 +++++++++++++++++++
 .../pypaimon/manifest/index_manifest_file.py  |  8 ++---
 .../manifest/manifest_file_manager.py         |  8 ++---
 .../manifest/manifest_list_manager.py         |  8 ++---
 .../read/reader/format_avro_reader.py         |  8 ++---
 5 files changed, 37 insertions(+), 24 deletions(-)

diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index 049d791fdc35..7b7206009616 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -228,6 +228,35 @@ run_zstd_manifest_test() {
 }
 # Main execution
 main() {
+    PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "unknown")
+    
+    # For Python 3.6, only run zstd manifest test
+    if [ "$PYTHON_VERSION" = "3.6" ]; then
+        echo -e "${YELLOW}Python 3.6 detected: Running only Zstd Manifest Test${NC}"
+        echo ""
+        
+        local zstd_manifest_result=0
+        if ! run_zstd_manifest_test; then
+            zstd_manifest_result=1
+        fi
+        
+        echo ""
+        echo -e "${YELLOW}=== Test Results Summary ===${NC}"
+        if [[ $zstd_manifest_result -eq 0 ]]; then
+            echo -e "${GREEN}✓ Zstd Manifest Test (Python 3.6 Compatibility): PASSED${NC}"
+            echo ""
+            cleanup_warehouse
+            echo -e "${GREEN}🎉 All tests passed!${NC}"
+            return 0
+        else
+            echo -e "${RED}✗ Zstd Manifest Test (Python 3.6 Compatibility): FAILED${NC}"
+            echo ""
+            cleanup_warehouse
+            return 1
+        fi
+    fi
+    
+    # For other Python versions, run all tests
     local java_write_result=0
     local python_read_result=0
     local python_write_result=0
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py
index 8c1770e207ed..d8bd33c8b337 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -20,17 +20,13 @@
 
 try:
     from pypaimon.manifest import fastavro_py36_compat
-except ImportError:
-    fastavro_py36_compat = None
-
-import fastavro
-
-try:
     if fastavro_py36_compat is not None:
         fastavro_py36_compat._apply_zstd_patch()
 except (ImportError, AttributeError, NameError):
     pass
 
+import fastavro
+
 from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
 from pypaimon.index.index_file_meta import IndexFileMeta
 from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index c7269146bb6c..075bba703576 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -21,17 +21,13 @@
 
 try:
     from pypaimon.manifest import fastavro_py36_compat
-except ImportError:
-    fastavro_py36_compat = None
-
-import fastavro
-
-try:
     if fastavro_py36_compat is not None:
         fastavro_py36_compat._apply_zstd_patch()
 except (ImportError, AttributeError, NameError):
     pass
 
+import fastavro
+
 from datetime import datetime
 
 from pypaimon.manifest.schema.data_file_meta import DataFileMeta
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index 99906a147607..3b10ddd8df88 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -21,17 +21,13 @@
 
 try:
     from pypaimon.manifest import fastavro_py36_compat
-except ImportError:
-    fastavro_py36_compat = None
-
-import fastavro
-
-try:
     if fastavro_py36_compat is not None:
         fastavro_py36_compat._apply_zstd_patch()
 except (ImportError, AttributeError, NameError):
     pass
 
+import fastavro
+
 from pypaimon.manifest.schema.manifest_file_meta import (
     MANIFEST_FILE_META_SCHEMA, ManifestFileMeta)
 from pypaimon.manifest.schema.simple_stats import SimpleStats
diff --git a/paimon-python/pypaimon/read/reader/format_avro_reader.py b/paimon-python/pypaimon/read/reader/format_avro_reader.py
index 32a9e283769e..2820eed79677 100644
--- a/paimon-python/pypaimon/read/reader/format_avro_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_avro_reader.py
@@ -20,16 +20,12 @@
 
 try:
     from pypaimon.manifest import fastavro_py36_compat
-except ImportError:
-    fastavro_py36_compat = None
-
-import fastavro
-
-try:
     if fastavro_py36_compat is not None:
         fastavro_py36_compat._apply_zstd_patch()
 except (ImportError, AttributeError, NameError):
     pass
+
+import fastavro
 import pyarrow as pa
 import pyarrow.dataset as ds
 from pyarrow import RecordBatch

From c79e64a01e05c363673a660f7befc5ee252b75ff Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 20:50:55 +0800
Subject: [PATCH 07/17] fix CI for py36

---
 paimon-python/pypaimon/__init__.py          | 9 +++++++++
 paimon-python/pypaimon/manifest/__init__.py | 9 +++++++++
 2 files changed, 18 insertions(+)

diff --git a/paimon-python/pypaimon/__init__.py b/paimon-python/pypaimon/__init__.py
index 5313e8e18aaf..ab080b0742b8 100644
--- a/paimon-python/pypaimon/__init__.py
+++ b/paimon-python/pypaimon/__init__.py
@@ -15,6 +15,15 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
+import sys
+if sys.version_info[:2] == (3, 6):
+    try:
+        from pypaimon.manifest import fastavro_py36_compat
+        if fastavro_py36_compat is not None:
+            fastavro_py36_compat._apply_zstd_patch()
+    except (ImportError, AttributeError, NameError):
+        pass
+
 from pypaimon.catalog.catalog_factory import CatalogFactory
 from pypaimon.filesystem.pvfs import PaimonVirtualFileSystem
 from pypaimon.schema.schema import Schema
diff --git a/paimon-python/pypaimon/manifest/__init__.py b/paimon-python/pypaimon/manifest/__init__.py
index 65b48d4d79b4..bb8ced23f53f 100644
--- a/paimon-python/pypaimon/manifest/__init__.py
+++ b/paimon-python/pypaimon/manifest/__init__.py
@@ -15,3 +15,12 @@
 #  See the License for the specific language governing permissions and
 # limitations under the License.
 ################################################################################
+
+import sys
+if sys.version_info[:2] == (3, 6):
+    try:
+        from pypaimon.manifest import fastavro_py36_compat
+        if fastavro_py36_compat is not None:
+            fastavro_py36_compat._apply_zstd_patch()
+    except (ImportError, AttributeError, NameError):
+        pass

From 8765b7c59f3e3db7e5fe224fdd3a22285e9e8389 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 21:16:47 +0800
Subject: [PATCH 08/17] add fastavro_py36_compat.py

---
 .../pypaimon/manifest/fastavro_py36_compat.py | 136 ++++++++++++++++++
 1 file changed, 136 insertions(+)
 create mode 100644 paimon-python/pypaimon/manifest/fastavro_py36_compat.py

diff --git a/paimon-python/pypaimon/manifest/fastavro_py36_compat.py b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
new file mode 100644
index 000000000000..93b9599721df
--- /dev/null
+++ b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
@@ -0,0 +1,136 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+################################################################################
+
+"""
+ Provides compatibility patches for fastavro on Python 3.6,
+specifically for handling zstd-compressed Avro files.
+
+The main issue addressed is:
+- On Python 3.6, fastavro's zstd decompression may fail with:
+  "zstd.ZstdError: could not determine content size in frame header"
+  
+This module patches fastavro's zstd handling to use a more compatible
+decompression method that works on Python 3.6.
+"""
+
+
+import sys
+
+_patch_applied = False
+
+
+def _apply_zstd_patch():
+    global _patch_applied
+    if _patch_applied:
+        return
+
+    if sys.version_info[:2] != (3, 6):
+        return
+
+    try:
+        import fastavro
+        import zstandard as zstd
+        from io import BytesIO
+
+        try:
+            import fastavro._read as fastavro_read
+        except (ImportError, AttributeError):
+            try:
+                fastavro_read = fastavro._read
+            except (AttributeError, ImportError):
+                return
+
+        def _fixed_zstandard_read_block(decoder):
+            from fastavro._read import read_long
+            length = read_long(decoder)
+            
+            try:
+                if hasattr(decoder, 'read_fixed'):
+                    data = decoder.read_fixed(length)
+                elif hasattr(decoder, 'read'):
+                    data = decoder.read(length)
+                else:
+                    from fastavro._read import read_fixed
+                    data = read_fixed(decoder, length)
+            except (TypeError, AttributeError):
+                if hasattr(decoder, 'read'):
+                    data = decoder.read(length)
+                else:
+                    raise
+
+            decompressor = zstd.ZstdDecompressor()
+            with decompressor.stream_reader(BytesIO(data)) as reader:
+                decompressed = reader.read()
+                return BytesIO(decompressed)
+
+        try:
+            if hasattr(fastavro_read, 'BLOCK_READERS'):
+                block_readers = fastavro_read.BLOCK_READERS
+
+                block_readers['zstandard'] = _fixed_zstandard_read_block
+                block_readers['zstd'] = _fixed_zstandard_read_block
+                
+                try:
+                    if hasattr(fastavro_read, '__dict__'):
+                        fastavro_read.__dict__['zstandard_read_block'] = _fixed_zstandard_read_block
+                except (TypeError, AttributeError):
+                    pass
+                
+                if block_readers.get('zstandard') is _fixed_zstandard_read_block:
+                    _patch_applied = True
+        except (TypeError, AttributeError) as e:
+            pass
+
+        if not _patch_applied:
+            try:
+                setattr(fastavro_read, 'zstandard_read_block', _fixed_zstandard_read_block)
+                if hasattr(fastavro, '_read') and hasattr(fastavro._read, 'zstandard_read_block'):
+                    setattr(fastavro._read, 'zstandard_read_block', _fixed_zstandard_read_block)
+                _patch_applied = True
+            except (TypeError, AttributeError):
+                pass
+
+        if not _patch_applied:
+            try:
+                import fastavro._read
+                if hasattr(fastavro._read, '__dict__'):
+                    fastavro._read.__dict__['zstandard_read_block'] = _fixed_zstandard_read_block
+                else:
+                    fastavro._read.zstandard_read_block = _fixed_zstandard_read_block
+                _patch_applied = True
+            except (TypeError, AttributeError):
+                pass
+
+    except (ImportError, AttributeError):
+        pass
+
+
+if sys.version_info[:2] == (3, 6):
+    try:
+        import fastavro
+        _apply_zstd_patch()
+        try:
+            import fastavro._read as fastavro_read
+            if hasattr(fastavro_read, 'BLOCK_READERS'):
+                block_readers = fastavro_read.BLOCK_READERS
+                if 'zstandard' in block_readers or 'zstd' in block_readers:
+                    pass
+        except (ImportError, AttributeError):
+            pass
+    except ImportError:
+        pass

From bc178436032e5805f99ec72e4f7567d8afc284f4 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 21:41:54 +0800
Subject: [PATCH 09/17] fix CI

---
 paimon-python/pypaimon/__init__.py                    |  2 +-
 paimon-python/pypaimon/manifest/__init__.py           |  6 +++++-
 .../pypaimon/manifest/fastavro_py36_compat.py         | 11 +----------
 3 files changed, 7 insertions(+), 12 deletions(-)

diff --git a/paimon-python/pypaimon/__init__.py b/paimon-python/pypaimon/__init__.py
index ab080b0742b8..9cb3539e6cff 100644
--- a/paimon-python/pypaimon/__init__.py
+++ b/paimon-python/pypaimon/__init__.py
@@ -18,7 +18,7 @@
 import sys
 if sys.version_info[:2] == (3, 6):
     try:
-        from pypaimon.manifest import fastavro_py36_compat
+        from pypaimon.manifest import fastavro_py36_compat  # noqa: F401
         if fastavro_py36_compat is not None:
             fastavro_py36_compat._apply_zstd_patch()
     except (ImportError, AttributeError, NameError):
diff --git a/paimon-python/pypaimon/manifest/__init__.py b/paimon-python/pypaimon/manifest/__init__.py
index bb8ced23f53f..e3dddf5b7765 100644
--- a/paimon-python/pypaimon/manifest/__init__.py
+++ b/paimon-python/pypaimon/manifest/__init__.py
@@ -16,11 +16,15 @@
 # limitations under the License.
 ################################################################################
 
+# Apply fastavro Python 3.6 compatibility patch early, before any other
+# manifest modules are imported that might use fastavro
 import sys
 if sys.version_info[:2] == (3, 6):
     try:
-        from pypaimon.manifest import fastavro_py36_compat
+        # Import fastavro_py36_compat module (this file is in the same directory)
+        from pypaimon.manifest import fastavro_py36_compat  # noqa: F401
         if fastavro_py36_compat is not None:
             fastavro_py36_compat._apply_zstd_patch()
     except (ImportError, AttributeError, NameError):
+        # Module may not be available in some environments, silently skip
         pass
diff --git a/paimon-python/pypaimon/manifest/fastavro_py36_compat.py b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
index 93b9599721df..7f5994b9614b 100644
--- a/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
+++ b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
@@ -93,7 +93,7 @@ def _fixed_zstandard_read_block(decoder):
                 
                 if block_readers.get('zstandard') is _fixed_zstandard_read_block:
                     _patch_applied = True
-        except (TypeError, AttributeError) as e:
+        except (TypeError, AttributeError):
             pass
 
         if not _patch_applied:
@@ -122,15 +122,6 @@ def _fixed_zstandard_read_block(decoder):
 
 if sys.version_info[:2] == (3, 6):
     try:
-        import fastavro
         _apply_zstd_patch()
-        try:
-            import fastavro._read as fastavro_read
-            if hasattr(fastavro_read, 'BLOCK_READERS'):
-                block_readers = fastavro_read.BLOCK_READERS
-                if 'zstandard' in block_readers or 'zstd' in block_readers:
-                    pass
-        except (ImportError, AttributeError):
-            pass
     except ImportError:
         pass

From fe07e0760d1d1019dc29606c59c7d5c1e6341b82 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 22:24:15 +0800
Subject: [PATCH 10/17] clean code

---
 paimon-python/pypaimon/__init__.py            |   4 +-
 paimon-python/pypaimon/manifest/__init__.py   |   6 +-
 .../pypaimon/manifest/fastavro_py36_compat.py | 106 +++++-------------
 .../pypaimon/manifest/index_manifest_file.py  |   7 --
 .../manifest/manifest_file_manager.py         |   7 --
 .../manifest/manifest_list_manager.py         |   7 --
 .../read/reader/format_avro_reader.py         |   7 --
 7 files changed, 30 insertions(+), 114 deletions(-)

diff --git a/paimon-python/pypaimon/__init__.py b/paimon-python/pypaimon/__init__.py
index 9cb3539e6cff..024f8b732bf3 100644
--- a/paimon-python/pypaimon/__init__.py
+++ b/paimon-python/pypaimon/__init__.py
@@ -19,9 +19,7 @@
 if sys.version_info[:2] == (3, 6):
     try:
         from pypaimon.manifest import fastavro_py36_compat  # noqa: F401
-        if fastavro_py36_compat is not None:
-            fastavro_py36_compat._apply_zstd_patch()
-    except (ImportError, AttributeError, NameError):
+    except ImportError:
         pass
 
 from pypaimon.catalog.catalog_factory import CatalogFactory
diff --git a/paimon-python/pypaimon/manifest/__init__.py b/paimon-python/pypaimon/manifest/__init__.py
index e3dddf5b7765..4b50145a36c5 100644
--- a/paimon-python/pypaimon/manifest/__init__.py
+++ b/paimon-python/pypaimon/manifest/__init__.py
@@ -21,10 +21,6 @@
 import sys
 if sys.version_info[:2] == (3, 6):
     try:
-        # Import fastavro_py36_compat module (this file is in the same directory)
         from pypaimon.manifest import fastavro_py36_compat  # noqa: F401
-        if fastavro_py36_compat is not None:
-            fastavro_py36_compat._apply_zstd_patch()
-    except (ImportError, AttributeError, NameError):
-        # Module may not be available in some environments, silently skip
+    except ImportError:
         pass
diff --git a/paimon-python/pypaimon/manifest/fastavro_py36_compat.py b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
index 7f5994b9614b..f6f509fc0ec5 100644
--- a/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
+++ b/paimon-python/pypaimon/manifest/fastavro_py36_compat.py
@@ -17,7 +17,7 @@
 ################################################################################
 
 """
- Provides compatibility patches for fastavro on Python 3.6,
+Provides compatibility patches for fastavro on Python 3.6,
 specifically for handling zstd-compressed Avro files.
 
 The main issue addressed is:
@@ -28,7 +28,6 @@
 decompression method that works on Python 3.6.
 """
 
-
 import sys
 
 _patch_applied = False
@@ -36,88 +35,39 @@
 
 def _apply_zstd_patch():
     global _patch_applied
-    if _patch_applied:
-        return
-
-    if sys.version_info[:2] != (3, 6):
+    if _patch_applied or sys.version_info[:2] != (3, 6):
         return
 
     try:
-        import fastavro
+        import fastavro._read as fastavro_read
         import zstandard as zstd
         from io import BytesIO
-
-        try:
-            import fastavro._read as fastavro_read
-        except (ImportError, AttributeError):
-            try:
-                fastavro_read = fastavro._read
-            except (AttributeError, ImportError):
-                return
-
-        def _fixed_zstandard_read_block(decoder):
-            from fastavro._read import read_long
-            length = read_long(decoder)
-            
-            try:
-                if hasattr(decoder, 'read_fixed'):
-                    data = decoder.read_fixed(length)
-                elif hasattr(decoder, 'read'):
-                    data = decoder.read(length)
-                else:
-                    from fastavro._read import read_fixed
-                    data = read_fixed(decoder, length)
-            except (TypeError, AttributeError):
-                if hasattr(decoder, 'read'):
-                    data = decoder.read(length)
-                else:
-                    raise
-
-            decompressor = zstd.ZstdDecompressor()
-            with decompressor.stream_reader(BytesIO(data)) as reader:
-                decompressed = reader.read()
-                return BytesIO(decompressed)
-
-        try:
-            if hasattr(fastavro_read, 'BLOCK_READERS'):
-                block_readers = fastavro_read.BLOCK_READERS
-
-                block_readers['zstandard'] = _fixed_zstandard_read_block
-                block_readers['zstd'] = _fixed_zstandard_read_block
-                
-                try:
-                    if hasattr(fastavro_read, '__dict__'):
-                        fastavro_read.__dict__['zstandard_read_block'] = _fixed_zstandard_read_block
-                except (TypeError, AttributeError):
-                    pass
-                
-                if block_readers.get('zstandard') is _fixed_zstandard_read_block:
-                    _patch_applied = True
-        except (TypeError, AttributeError):
-            pass
-
-        if not _patch_applied:
-            try:
-                setattr(fastavro_read, 'zstandard_read_block', _fixed_zstandard_read_block)
-                if hasattr(fastavro, '_read') and hasattr(fastavro._read, 'zstandard_read_block'):
-                    setattr(fastavro._read, 'zstandard_read_block', _fixed_zstandard_read_block)
-                _patch_applied = True
-            except (TypeError, AttributeError):
-                pass
-
-        if not _patch_applied:
-            try:
-                import fastavro._read
-                if hasattr(fastavro._read, '__dict__'):
-                    fastavro._read.__dict__['zstandard_read_block'] = _fixed_zstandard_read_block
-                else:
-                    fastavro._read.zstandard_read_block = _fixed_zstandard_read_block
-                _patch_applied = True
-            except (TypeError, AttributeError):
-                pass
-
     except (ImportError, AttributeError):
-        pass
+        return
+
+    def _fixed_zstandard_read_block(decoder):
+        from fastavro._read import read_long
+        
+        length = read_long(decoder)
+        
+        if hasattr(decoder, 'read_fixed'):
+            data = decoder.read_fixed(length)
+        elif hasattr(decoder, 'read'):
+            data = decoder.read(length)
+        else:
+            from fastavro._read import read_fixed
+            data = read_fixed(decoder, length)
+
+        decompressor = zstd.ZstdDecompressor()
+        with decompressor.stream_reader(BytesIO(data)) as reader:
+            decompressed = reader.read()
+            return BytesIO(decompressed)
+
+    if hasattr(fastavro_read, 'BLOCK_READERS'):
+        block_readers = fastavro_read.BLOCK_READERS
+        block_readers['zstandard'] = _fixed_zstandard_read_block
+        block_readers['zstd'] = _fixed_zstandard_read_block
+        _patch_applied = True
 
 
 if sys.version_info[:2] == (3, 6):
diff --git a/paimon-python/pypaimon/manifest/index_manifest_file.py b/paimon-python/pypaimon/manifest/index_manifest_file.py
index d8bd33c8b337..cd66ee5fce7e 100644
--- a/paimon-python/pypaimon/manifest/index_manifest_file.py
+++ b/paimon-python/pypaimon/manifest/index_manifest_file.py
@@ -18,13 +18,6 @@
 from io import BytesIO
 from typing import List
 
-try:
-    from pypaimon.manifest import fastavro_py36_compat
-    if fastavro_py36_compat is not None:
-        fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError, NameError):
-    pass
-
 import fastavro
 
 from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
diff --git a/paimon-python/pypaimon/manifest/manifest_file_manager.py b/paimon-python/pypaimon/manifest/manifest_file_manager.py
index 075bba703576..f6ae41e3d386 100644
--- a/paimon-python/pypaimon/manifest/manifest_file_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_file_manager.py
@@ -19,13 +19,6 @@
 from io import BytesIO
 from typing import List
 
-try:
-    from pypaimon.manifest import fastavro_py36_compat
-    if fastavro_py36_compat is not None:
-        fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError, NameError):
-    pass
-
 import fastavro
 
 from datetime import datetime
diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index 3b10ddd8df88..e4331408bddf 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -19,13 +19,6 @@
 from io import BytesIO
 from typing import List
 
-try:
-    from pypaimon.manifest import fastavro_py36_compat
-    if fastavro_py36_compat is not None:
-        fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError, NameError):
-    pass
-
 import fastavro
 
 from pypaimon.manifest.schema.manifest_file_meta import (
diff --git a/paimon-python/pypaimon/read/reader/format_avro_reader.py b/paimon-python/pypaimon/read/reader/format_avro_reader.py
index 2820eed79677..4114d8e93b59 100644
--- a/paimon-python/pypaimon/read/reader/format_avro_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_avro_reader.py
@@ -18,13 +18,6 @@
 
 from typing import List, Optional, Any
 
-try:
-    from pypaimon.manifest import fastavro_py36_compat
-    if fastavro_py36_compat is not None:
-        fastavro_py36_compat._apply_zstd_patch()
-except (ImportError, AttributeError, NameError):
-    pass
-
 import fastavro
 import pyarrow as pa
 import pyarrow.dataset as ds

From 36cd0607ab0a4414957836e0520e3f11f2c7f022 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Thu, 8 Jan 2026 22:33:34 +0800
Subject: [PATCH 11/17] clean code

---
 paimon-python/pypaimon/manifest/manifest_list_manager.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py
index e4331408bddf..a1897fbee721 100644
--- a/paimon-python/pypaimon/manifest/manifest_list_manager.py
+++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py
@@ -59,7 +59,6 @@ def read(self, manifest_list_name: str) -> List[ManifestFileMeta]:
             avro_bytes = input_stream.read()
         buffer = BytesIO(avro_bytes)
         reader = fastavro.reader(buffer)
-        
         for record in reader:
             stats_dict = dict(record['_PARTITION_STATS'])
             partition_stats = SimpleStats(

From 2d3a43e06dc09008f55aef29b775b55a7f530d65 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 00:30:57 +0800
Subject: [PATCH 12/17] remove extra case

---
 .../java/org/apache/paimon/JavaPyE2ETest.java | 34 --------
 paimon-python/dev/run_mixed_tests.sh          | 77 +------------------
 .../tests/e2e/java_py_read_write_test.py      | 36 +++------
 .../write/writer/key_value_data_writer.py     |  4 +-
 4 files changed, 14 insertions(+), 137 deletions(-)

diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 907b263a7138..39c2011bf486 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -67,7 +67,6 @@
 
 import static org.apache.paimon.CoreOptions.BUCKET;
 import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
-import static org.apache.paimon.CoreOptions.MANIFEST_COMPRESSION;
 import static org.apache.paimon.CoreOptions.TARGET_FILE_SIZE;
 import static org.apache.paimon.data.DataFormatTestUtil.internalRowToString;
 import static org.apache.paimon.table.SimpleTableTestBase.getResult;
@@ -380,39 +379,6 @@ public void testReadPkTable() throws Exception {
                         "6, Beef, Meat, 8.0");
     }
 
-    @Test
-    @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
-    public void testJavaWriteZstdManifestTable() throws Exception {
-        Identifier identifier = identifier("zstd_manifest_test_table");
-
-        Schema schema =
-                Schema.newBuilder()
-                        .column("id", DataTypes.INT())
-                        .column("name", DataTypes.STRING())
-                        .column("value", DataTypes.DOUBLE())
-                        .option(MANIFEST_COMPRESSION.key(), "zstd")
-                        .build();
-
-        catalog.createTable(identifier, schema, true);
-        Table table = catalog.getTable(identifier);
-        FileStoreTable fileStoreTable = (FileStoreTable) table;
-
-        String manifestCompression = fileStoreTable.coreOptions().manifestCompression();
-        assertThat(manifestCompression).isEqualTo("zstd");
-
-        try (StreamTableWrite write = fileStoreTable.newWrite(commitUser);
-                InnerTableCommit commit = fileStoreTable.newCommit(commitUser)) {
-
-            write.write(GenericRow.of(1, BinaryString.fromString("test1"), 1.1));
-            write.write(GenericRow.of(2, BinaryString.fromString("test2"), 2.2));
-            write.write(GenericRow.of(3, BinaryString.fromString("test3"), 3.3));
-
-            commit.commit(0, write.prepareCommit(true, 0));
-        }
-
-        assertThat(fileStoreTable.newSnapshotReader().read().dataSplits()).hasSizeGreaterThan(0);
-    }
-
     // Helper method from TableTestBase
     protected Identifier identifier(String tableName) {
         return new Identifier(database, tableName);
diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index 7b7206009616..a93ee19ac446 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -193,76 +193,13 @@ run_pk_dv_test() {
     fi
 }
 
-# Function to run zstd manifest e2e test
-run_zstd_manifest_test() {
-    echo -e "${YELLOW}=== Step 6: Running Zstd Manifest Compression Test (Python 3.6 Compatibility) ===${NC}"
-
-    cd "$PROJECT_ROOT"
-
-    echo "Running Maven test for JavaPyE2ETest.testJavaWriteZstdManifestTable..."
-    local java_result=0
-    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testJavaWriteZstdManifestTable -pl paimon-core -Drun.e2e.tests=true; then
-        echo -e "${GREEN}✓ Java zstd manifest test completed successfully${NC}"
-    else
-        echo -e "${RED}✗ Java zstd manifest test failed${NC}"
-        java_result=1
-    fi
-
-    echo ""
-
-    cd "$PAIMON_PYTHON_DIR"
-    echo "Running Python test for JavaPyReadWriteTest.test_read_zstd_manifest_table..."
-    local python_result=0
-    if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest::test_read_zstd_manifest_table -v; then
-        echo -e "${GREEN}✓ Python zstd manifest test completed successfully${NC}"
-    else
-        echo -e "${RED}✗ Python zstd manifest test failed${NC}"
-        python_result=1
-    fi
-
-    if [[ $java_result -eq 0 && $python_result -eq 0 ]]; then
-        return 0
-    else
-        return 1
-    fi
-}
 # Main execution
 main() {
-    PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "unknown")
-    
-    # For Python 3.6, only run zstd manifest test
-    if [ "$PYTHON_VERSION" = "3.6" ]; then
-        echo -e "${YELLOW}Python 3.6 detected: Running only Zstd Manifest Test${NC}"
-        echo ""
-        
-        local zstd_manifest_result=0
-        if ! run_zstd_manifest_test; then
-            zstd_manifest_result=1
-        fi
-        
-        echo ""
-        echo -e "${YELLOW}=== Test Results Summary ===${NC}"
-        if [[ $zstd_manifest_result -eq 0 ]]; then
-            echo -e "${GREEN}✓ Zstd Manifest Test (Python 3.6 Compatibility): PASSED${NC}"
-            echo ""
-            cleanup_warehouse
-            echo -e "${GREEN}🎉 All tests passed!${NC}"
-            return 0
-        else
-            echo -e "${RED}✗ Zstd Manifest Test (Python 3.6 Compatibility): FAILED${NC}"
-            echo ""
-            cleanup_warehouse
-            return 1
-        fi
-    fi
-    
-    # For other Python versions, run all tests
     local java_write_result=0
     local python_read_result=0
     local python_write_result=0
     local java_read_result=0
     local pk_dv_result=0
-    local zstd_manifest_result=0
 
     echo -e "${YELLOW}Starting mixed language test execution...${NC}"
     echo ""
@@ -305,12 +242,6 @@ main() {
     if ! run_pk_dv_test; then
         pk_dv_result=1
     fi
-
-    # Run zstd manifest test
-    if ! run_zstd_manifest_test; then
-        zstd_manifest_result=1
-    fi
-
     echo ""
     echo -e "${YELLOW}=== Test Results Summary ===${NC}"
 
@@ -344,18 +275,12 @@ main() {
         echo -e "${RED}✗ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): FAILED${NC}"
     fi
 
-    if [[ $zstd_manifest_result -eq 0 ]]; then
-        echo -e "${GREEN}✓ Zstd Manifest Test (Python 3.6 Compatibility): PASSED${NC}"
-    else
-        echo -e "${RED}✗ Zstd Manifest Test (Python 3.6 Compatibility): FAILED${NC}"
-    fi
-
     echo ""
 
     # Clean up warehouse directory after all tests
     cleanup_warehouse
 
-    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 && $zstd_manifest_result -eq 0 ]]; then
+    if [[ $java_write_result -eq 0 && $python_read_result -eq 0 && $python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 ]]; then
         echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability verified.${NC}"
         return 0
     else
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 6a24dd1a4ce7..6a775db8ff1a 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -17,6 +17,7 @@
 ################################################################################
 
 import os
+import sys
 import unittest
 
 import pandas as pd
@@ -25,6 +26,12 @@
 from pypaimon.catalog.catalog_factory import CatalogFactory
 from pypaimon.schema.schema import Schema
 
+if sys.version_info[:2] == (3, 6):
+    from pypaimon.tests.py36.pyarrow_compat import table_sort_by
+else:
+    def table_sort_by(table: pa.Table, column_name: str, order: str = 'ascending') -> pa.Table:
+        return table.sort_by([(column_name, order)])
+
 
 class JavaPyReadWriteTest(unittest.TestCase):
     @classmethod
@@ -196,7 +203,7 @@ def test_pk_dv_read(self):
         read_builder = table.new_read_builder()
         table_read = read_builder.new_read()
         splits = read_builder.new_scan().plan().splits()
-        actual = table_read.to_arrow(splits).sort_by('pt')
+        actual = table_sort_by(table_read.to_arrow(splits), 'pt')
         expected = pa.Table.from_pydict({
             'pt': [1, 2, 2],
             'a': [10, 21, 22],
@@ -219,7 +226,7 @@ def test_pk_dv_read_multi_batch(self):
         read_builder = table.new_read_builder()
         table_read = read_builder.new_read()
         splits = read_builder.new_scan().plan().splits()
-        actual = table_read.to_arrow(splits).sort_by('pt')
+        actual = table_sort_by(table_read.to_arrow(splits), 'pt')
         expected = pa.Table.from_pydict({
             'pt': [1] * 9999,
             'a': [i * 10 for i in range(1, 10001) if i * 10 != 81930],
@@ -242,7 +249,7 @@ def test_pk_dv_read_multi_batch_raw_convertable(self):
         read_builder = table.new_read_builder()
         table_read = read_builder.new_read()
         splits = read_builder.new_scan().plan().splits()
-        actual = table_read.to_arrow(splits).sort_by('pt')
+        actual = table_sort_by(table_read.to_arrow(splits), 'pt')
         expected = pa.Table.from_pydict({
             'pt': [1] * 9999,
             'a': [i * 10 for i in range(1, 10001) if i * 10 != 81930],
@@ -250,26 +257,3 @@ def test_pk_dv_read_multi_batch_raw_convertable(self):
         }, schema=pa_schema)
         self.assertEqual(expected, actual)
 
-    def test_read_zstd_manifest_table(self):
-        table_name = 'default.zstd_manifest_test_table'
-        
-        try:
-            table = self.catalog.get_table(table_name)
-        except Exception as e:
-            self.fail(
-                f"Failed to get table {table_name}. "
-                f"Make sure Java test (JavaPyE2ETest.testJavaWriteZstdManifestTable) "
-                f"has been run first. Error: {e}"
-            )
-
-        read_builder = table.new_read_builder()
-        table_scan = read_builder.new_scan()
-        table_read = read_builder.new_read()
-        
-        splits = table_scan.plan().splits()
-        result = table_read.to_pandas(splits)
-        
-        self.assertEqual(len(result), 3)
-        expected_ids = {1, 2, 3}
-        actual_ids = set(result['id'].tolist())
-        self.assertEqual(actual_ids, expected_ids)
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index 5e82369b6c97..36f26dcd5f2b 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -38,8 +38,10 @@ def _add_system_fields(self, data: pa.RecordBatch) -> pa.RecordBatch:
         num_rows = data.num_rows
         enhanced_table = data
 
+        column_names = data.schema.names
+
         for pk_key in reversed(self.trimmed_primary_keys):
-            if pk_key in data.column_names:
+            if pk_key in column_names:
                 key_column = data.column(pk_key)
                 enhanced_table = enhanced_table.add_column(0, f'_KEY_{pk_key}', key_column)
 

From 515ab4d55943a3a9f87ed4ec3c7f01506c465f47 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 00:47:24 +0800
Subject: [PATCH 13/17] fix

---
 paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 6a775db8ff1a..d2929d402c08 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -255,5 +255,4 @@ def test_pk_dv_read_multi_batch_raw_convertable(self):
             'a': [i * 10 for i in range(1, 10001) if i * 10 != 81930],
             'b': [i * 100 for i in range(1, 10001) if i * 10 != 81930]
         }, schema=pa_schema)
-        self.assertEqual(expected, actual)
-
+        self.assertEqual(expected, actual)
\ No newline at end of file

From 415f0878961c6ca43a4b7bbd5b7376f2137999df Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 07:21:15 +0800
Subject: [PATCH 14/17] fix

---
 paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index d2929d402c08..f1b5cbe234d3 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -255,4 +255,4 @@ def test_pk_dv_read_multi_batch_raw_convertable(self):
             'a': [i * 10 for i in range(1, 10001) if i * 10 != 81930],
             'b': [i * 100 for i in range(1, 10001) if i * 10 != 81930]
         }, schema=pa_schema)
-        self.assertEqual(expected, actual)
\ No newline at end of file
+        self.assertEqual(expected, actual)

From 823cf8068aa16739fc69f98aa9b68ad851b39dbb Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 08:13:04 +0800
Subject: [PATCH 15/17] fix

---
 .../java/org/apache/paimon/JavaPyE2ETest.java |  2 ++
 .../org/apache/paimon/JavaPyLanceE2ETest.java | 13 ++---------
 paimon-python/dev/run_mixed_tests.sh          |  7 ++++--
 .../tests/e2e/java_py_read_write_test.py      | 22 ++++++++++++-------
 4 files changed, 23 insertions(+), 21 deletions(-)

diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 39c2011bf486..03642b0e5547 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -53,6 +53,7 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
 import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
 
 import java.nio.file.Files;
@@ -356,6 +357,7 @@ public void testPKDeletionVectorWriteMultiBatchRawConvertable() throws Exception
 
     @Test
     @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    @DisabledIfSystemProperty(named = "python.version", matches = "3.6")
     public void testReadPkTable() throws Exception {
         Identifier identifier = identifier("mixed_test_pk_tablep_parquet");
         Table table = catalog.getTable(identifier);
diff --git a/paimon-lance/src/test/java/org/apache/paimon/JavaPyLanceE2ETest.java b/paimon-lance/src/test/java/org/apache/paimon/JavaPyLanceE2ETest.java
index 290193340a42..a431657b95e4 100644
--- a/paimon-lance/src/test/java/org/apache/paimon/JavaPyLanceE2ETest.java
+++ b/paimon-lance/src/test/java/org/apache/paimon/JavaPyLanceE2ETest.java
@@ -51,6 +51,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
 import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
 
 import java.nio.file.Files;
@@ -260,19 +261,9 @@ public void testJavaWriteReadPkTableLance() throws Exception {
 
     @Test
     @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+    @DisabledIfSystemProperty(named = "python.version", matches = "3.6")
     public void testReadPkTableLance() throws Exception {
         try {
-            // Known issue: Reading Python-written Lance files in Java causes JVM crash due to
-            // missing Tokio runtime. The error is:
-            // "there is no reactor running, must be called from the context of a Tokio 1.x runtime"
-            //
-            // This is a limitation of lance-core Java bindings. The Rust native library requires
-            // Tokio runtime for certain operations when reading files written by Python (which may
-            // use different encoding formats). Java-written files can be read successfully because
-            // they use synchronous APIs that don't require Tokio.
-            //
-            // Workaround: Try to "warm up" Tokio runtime by reading a Java-written file first.
-            // This may initialize the Tokio runtime if it's created on first use.
             try {
                 Identifier warmupIdentifier = identifier("mixed_test_pk_tablej_lance");
                 try {
diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index a93ee19ac446..d8e9eac969c1 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -138,11 +138,14 @@ run_java_read_test() {
 
     cd "$PROJECT_ROOT"
 
+    PYTHON_VERSION=$(python -c "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')" 2>/dev/null || echo "unknown")
+    echo "Detected Python version: $PYTHON_VERSION"
+
     # Run Java test for parquet format in paimon-core
     echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read Parquet)..."
     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 -pl paimon-core -Drun.e2e.tests=true; then
+    if mvn test -Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable -pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
         echo -e "${GREEN}✓ Java read parquet test completed successfully${NC}"
     else
         echo -e "${RED}✗ Java read parquet test failed${NC}"
@@ -155,7 +158,7 @@ run_java_read_test() {
     echo "Running Maven test for JavaPyLanceE2ETest.testReadPkTableLance (Java Read Lance)..."
     echo "Note: Maven may download dependencies on first run, this may take a while..."
     local lance_result=0
-    if mvn test -Dtest=org.apache.paimon.JavaPyLanceE2ETest#testReadPkTableLance -pl paimon-lance -Drun.e2e.tests=true; then
+    if mvn test -Dtest=org.apache.paimon.JavaPyLanceE2ETest#testReadPkTableLance -pl paimon-lance -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
         echo -e "${GREEN}✓ Java read lance test completed successfully${NC}"
     else
         echo -e "${RED}✗ Java read lance test failed${NC}"
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index f1b5cbe234d3..7a194574f5e5 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -33,6 +33,13 @@ def table_sort_by(table: pa.Table, column_name: str, order: str = 'ascending') -
         return table.sort_by([(column_name, order)])
 
 
+def get_file_format_params():
+    if sys.version_info[:2] == (3, 6):
+        return [('parquet',)]
+    else:
+        return [('parquet',), ('lance',)]
+
+
 class JavaPyReadWriteTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -96,11 +103,13 @@ def test_read_append_table(self):
         res = table_read.to_pandas(table_scan.plan().splits())
         print(res)
 
-    @parameterized.expand([
-        ('parquet',),
-        ('lance',),
-    ])
+    @parameterized.expand(get_file_format_params())
     def test_py_write_read_pk_table(self, file_format):
+        if sys.version_info[:2] == (3, 6):
+            self.skipTest(
+                "Skipping on Python 3.6 due to PyArrow compatibility issue (RecordBatch.add_column not available). "
+                "Will be fixed in next PR."
+            )
         pa_schema = pa.schema([
             ('id', pa.int32()),
             ('name', pa.string()),
@@ -157,10 +166,7 @@ def test_py_write_read_pk_table(self, file_format):
         actual_names = set(initial_result['name'].tolist())
         self.assertEqual(actual_names, expected_names)
 
-    @parameterized.expand([
-        ('parquet',),
-        ('lance',),
-    ])
+    @parameterized.expand(get_file_format_params())
     def test_read_pk_table(self, file_format):
         # For parquet, read from Java-written table (no format suffix)
         # For lance, read from Java-written table (with format suffix)

From 21e4b86beef00fffc4128194175811d796ef3d28 Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 12:54:47 +0800
Subject: [PATCH 16/17] fix format

---
 paimon-python/dev/run_mixed_tests.sh | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/paimon-python/dev/run_mixed_tests.sh b/paimon-python/dev/run_mixed_tests.sh
index d8e9eac969c1..38a0ada6c228 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -195,7 +195,6 @@ run_pk_dv_test() {
         return 1
     fi
 }
-
 # Main execution
 main() {
     local java_write_result=0
@@ -272,11 +271,11 @@ main() {
         echo -e "${RED}✗ Java Read Test (Parquet + Lance): FAILED${NC}"
     fi
 
-    if [[ $pk_dv_result -eq 0 ]]; then
-        echo -e "${GREEN}✓ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): PASSED${NC}"
-    else
-        echo -e "${RED}✗ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): FAILED${NC}"
-    fi
+     if [[ $pk_dv_result -eq 0 ]]; then
+          echo -e "${GREEN}✓ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): PASSED${NC}"
+      else
+          echo -e "${RED}✗ PK DV Test (JavaPyReadWriteTest.testPKDeletionVectorWriteRead): FAILED${NC}"
+      fi
 
     echo ""
 

From 3a2bddaa701a43d38241b57e95750f4b3c35579f Mon Sep 17 00:00:00 2001
From: xiaohongbo <xiaohongbo.xhb@alibaba-inc.com>
Date: Fri, 9 Jan 2026 12:59:33 +0800
Subject: [PATCH 17/17] revert fix

---
 paimon-python/pypaimon/write/writer/key_value_data_writer.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index 36f26dcd5f2b..5e82369b6c97 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -38,10 +38,8 @@ def _add_system_fields(self, data: pa.RecordBatch) -> pa.RecordBatch:
         num_rows = data.num_rows
         enhanced_table = data
 
-        column_names = data.schema.names
-
         for pk_key in reversed(self.trimmed_primary_keys):
-            if pk_key in column_names:
+            if pk_key in data.column_names:
                 key_column = data.column(pk_key)
                 enhanced_table = enhanced_table.add_column(0, f'_KEY_{pk_key}', key_column)