(seatunnel) branch dev updated: [Fix][Connector-V2][InfluxDB] Fix NPE when query returns empty result (#11966)

[github-bot]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new 1491a8e0ad [Fix][Connector-V2][InfluxDB] Fix NPE when query returns 
empty result (#11966)
1491a8e0ad is described below

commit 1491a8e0ad6f0f77ed4853fd4db3d4ff1cdfbd2e
Author: yuge1805 <45219837+yuge1...@users.noreply.github.com>
AuthorDate: Wed Aug 26 08:13:16 2026 +0000

    [Fix][Connector-V2][InfluxDB] Fix NPE when query returns empty result 
(#11966)
---
 .../seatunnel/influxdb/source/InfluxDBSource.java  | 30 +++++++++++++++++++++-
 .../influxdb/source/InfluxdbSourceReader.java      |  7 ++++-
 2 files changed, 35 insertions(+), 2 deletions(-)

diff --git 
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSource.java
 
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSource.java
index 282880482e..4273ac4cf9 100644
--- 
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSource.java
+++ 
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSource.java
@@ -31,6 +31,8 @@ import 
org.apache.seatunnel.connectors.seatunnel.influxdb.exception.InfluxdbConn
 import 
org.apache.seatunnel.connectors.seatunnel.influxdb.exception.InfluxdbConnectorException;
 import 
org.apache.seatunnel.connectors.seatunnel.influxdb.state.InfluxDBSourceState;
 
+import org.apache.commons.collections4.CollectionUtils;
+
 import org.influxdb.InfluxDB;
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
@@ -112,7 +114,20 @@ public class InfluxDBSource
         try {
             QueryResult queryResult = influxdb.query(new Query(query, 
sourceConfig.getDatabase()));
 
-            List<QueryResult.Series> serieList = 
queryResult.getResults().get(0).getSeries();
+            List<QueryResult.Result> results = queryResult.getResults();
+            if (CollectionUtils.isEmpty(results)) {
+                log.warn(
+                        "InfluxDB query returned empty results, using default 
column index mapping.");
+                return buildDefaultColumnsIndex();
+            }
+
+            List<QueryResult.Series> serieList = results.get(0).getSeries();
+            if (CollectionUtils.isEmpty(serieList)) {
+                log.warn(
+                        "InfluxDB query returned no series (empty data), using 
default column index mapping.");
+                return buildDefaultColumnsIndex();
+            }
+
             List<String> fieldNames = new 
ArrayList<>(serieList.get(0).getColumns());
 
             return 
Arrays.stream(catalogTable.getSeaTunnelRowType().getFieldNames())
@@ -126,6 +141,19 @@ public class InfluxDBSource
         }
     }
 
+    /**
+     * Builds a default column index list based on the catalog table schema. 
Used when the query
+     * returns no data, where actual column indices are not needed since no 
rows will be read.
+     */
+    private List<Integer> buildDefaultColumnsIndex() {
+        int fieldCount = catalogTable.getSeaTunnelRowType().getTotalFields();
+        List<Integer> defaultIndexList = new ArrayList<>(fieldCount);
+        for (int i = 0; i < fieldCount; i++) {
+            defaultIndexList.add(i);
+        }
+        return defaultIndexList;
+    }
+
     private static int containTzFunction(String sql) {
         Pattern pattern = Pattern.compile("tz\\(.*\\)");
         Matcher matcher = pattern.matcher(sql);
diff --git 
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxdbSourceReader.java
 
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxdbSourceReader.java
index 16eec22839..9d59c3d659 100644
--- 
a/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxdbSourceReader.java
+++ 
b/seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxdbSourceReader.java
@@ -135,7 +135,12 @@ public class InfluxdbSourceReader implements 
SourceReader<SeaTunnelRow, InfluxDB
 
     private void read(InfluxDBSourceSplit split, Collector<SeaTunnelRow> 
output) {
         QueryResult queryResult = influxdb.query(new Query(split.getQuery(), 
config.getDatabase()));
-        for (QueryResult.Result result : queryResult.getResults()) {
+        List<QueryResult.Result> results = queryResult.getResults();
+        if (CollectionUtils.isEmpty(results)) {
+            log.debug("split[{}] reader influxDB query result is empty.", 
split.splitId());
+            return;
+        }
+        for (QueryResult.Result result : results) {
             List<QueryResult.Series> serieList = result.getSeries();
             if (CollectionUtils.isNotEmpty(serieList)) {
                 for (QueryResult.Series series : serieList) {