This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11516-1ce2d781566564d5bd603710e7442bd88c531414 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 2dea151bc1dfe010c753d78640423bc33a3a6c67 Author: FatMan <[email protected]> AuthorDate: Tue Oct 20 11:59:05 2026 +0000 [Fix][Connector-V2] Fix NPE when reading Parquet source on Spark (missing tableId) (#11516) Co-authored-by: chen <[email protected]> --- .../file/source/reader/ParquetReadStrategy.java | 2 +- .../source/reader/ParquetReadStrategyTest.java | 29 +++++++++++++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java index baa632495c..79246a14b5 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java @@ -97,7 +97,7 @@ public class ParquetReadStrategy extends AbstractReadStrategy { @Override public void read(String path, String tableId, Collector<SeaTunnelRow> output) throws FileConnectorException, IOException { - this.read(new FileSourceSplit(path), output); + this.read(new FileSourceSplit(tableId, path), output); } @Override diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategyTest.java b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategyTest.java index d6d38f0e26..70ac9f9240 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategyTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategyTest.java @@ -96,7 +96,31 @@ public class ParquetReadStrategyTest { Assertions.assertNotNull(seaTunnelRowTypeInfo); log.info(seaTunnelRowTypeInfo.toString()); TestCollector testCollector = new TestCollector(); + // BaseFileSourceReader reads single-table splits with an empty table id parquetReadStrategy.read(path, "", testCollector); + List<SeaTunnelRow> rows = testCollector.getRows(); + Assertions.assertFalse(rows.isEmpty()); + rows.forEach(row -> Assertions.assertEquals("", row.getTableId())); + } + + @Test + public void testParquetReadPreservesTableId() throws Exception { + URL resource = ParquetReadStrategyTest.class.getResource("/timestamp_as_int64.parquet"); + Assertions.assertNotNull(resource); + String path = Paths.get(resource.toURI()).toString(); + ParquetReadStrategy parquetReadStrategy = new ParquetReadStrategy(); + LocalFileSystemConf.LocalConf localConf = + new LocalFileSystemConf.LocalConf(FS_DEFAULT_NAME_DEFAULT); + parquetReadStrategy.init(localConf); + SeaTunnelRowType seaTunnelRowTypeInfo = parquetReadStrategy.getSeaTunnelRowTypeInfo(path); + Assertions.assertNotNull(seaTunnelRowTypeInfo); + // Multi-table readers (e.g. MultipleTableHiveSourceReader) pass the table path of the split + String tableId = "my_db.my_table"; + TestCollector testCollector = new TestCollector(); + parquetReadStrategy.read(path, tableId, testCollector); + List<SeaTunnelRow> rows = testCollector.getRows(); + Assertions.assertFalse(rows.isEmpty()); + rows.forEach(row -> Assertions.assertEquals(tableId, row.getTableId())); } @Test @@ -264,14 +288,17 @@ public class ParquetReadStrategyTest { Assertions.assertEquals("job_blue-collar", rowType.getFieldName(1)); TestCollector testCollector = new TestCollector(); + // The Avro-incompatible column name forces the native parquet fallback reader + String tableId = "my_db.my_table"; parquetReadStrategy.read( NativeParquetWriterWithAvroIncompatibleColumn.DATA_FILE_PATH, - "", + tableId, testCollector); Assertions.assertEquals(1, testCollector.getRows().size()); SeaTunnelRow row = testCollector.getRows().get(0); Assertions.assertEquals(1, row.getField(0)); Assertions.assertEquals("engineer", row.getField(1)); + Assertions.assertEquals(tableId, row.getTableId()); } finally { NativeParquetWriterWithAvroIncompatibleColumn.deleteFile(); }
