fix: 添加reader配置中column数组长度和查询结果列长度的校验

gk0916 · kanata163 · commit bae47099e053 · 2020-07-13T16:32:44.000+08:00
diff --git a/flinkx-clickhouse/flinkx-clickhouse-reader/src/main/java/com/dtstack/flinkx/clickhouse/format/ClickhouseInputFormat.java b/flinkx-clickhouse/flinkx-clickhouse-reader/src/main/java/com/dtstack/flinkx/clickhouse/format/ClickhouseInputFormat.java
@@ -72,7 +72,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             hasNext = resultSet.next();
 
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
diff --git a/flinkx-gbase/flinkx-gbase-reader/src/main/java/com/dtstack/flinkx/gbase/format/GbaseInputFormat.java b/flinkx-gbase/flinkx-gbase-reader/src/main/java/com/dtstack/flinkx/gbase/format/GbaseInputFormat.java
@@ -68,7 +68,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
 
         } catch (SQLException se) {
diff --git a/flinkx-mysql/flinkx-mysql-reader/src/main/java/com/dtstack/flinkx/mysql/format/MysqlInputFormat.java b/flinkx-mysql/flinkx-mysql-reader/src/main/java/com/dtstack/flinkx/mysql/format/MysqlInputFormat.java
@@ -72,7 +72,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
 
         } catch (SQLException se) {
diff --git a/flinkx-phoenix/flinkx-phoenix-reader/src/main/java/com/dtstack/flinkx/phoenix/format/PhoenixInputFormat.java b/flinkx-phoenix/flinkx-phoenix-reader/src/main/java/com/dtstack/flinkx/phoenix/format/PhoenixInputFormat.java
@@ -78,7 +78,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             hasNext = resultSet.next();
 
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
diff --git a/flinkx-postgresql/flinkx-postgresql-reader/src/main/java/com/dtstack/flinkx/postgresql/format/PostgresqlInputFormat.java b/flinkx-postgresql/flinkx-postgresql-reader/src/main/java/com/dtstack/flinkx/postgresql/format/PostgresqlInputFormat.java
@@ -68,7 +68,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
 
         } catch (SQLException se) {
diff --git a/flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java b/flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java
@@ -186,7 +186,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if (splitWithRowCol) {
                 columnCount = columnCount - 1;
             }
-
+            checkSize(columnCount, metaColumns);
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
 
         } catch (SQLException se) {
@@ -832,4 +832,17 @@ private void queryStartLocation() throws SQLException{
         }
     }
 
+    /**
+     * 校验columnCount和metaColumns的长度是否相等
+     * @param columnCount
+     * @param metaColumns
+     */
+    protected void checkSize(int columnCount, List<MetaColumn> metaColumns) {
+        if (!ConstantValue.STAR_SYMBOL.equals(metaColumns.get(0).getName()) && columnCount != metaColumns.size()) {
+            LOG.error("error config: column size for reader is {},but columns size for query result is {}." +
+                            " And the query sql is '{}'.",
+                    metaColumns.size(), columnCount, querySql);
+            throw new RuntimeException("");
+        }
+    }
 }
diff --git a/flinkx-saphana/flinkx-saphana-reader/src/main/java/com/dtstack/flinkx/saphana/format/SaphanaInputFormat.java b/flinkx-saphana/flinkx-saphana-reader/src/main/java/com/dtstack/flinkx/saphana/format/SaphanaInputFormat.java
@@ -76,7 +76,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             hasNext = resultSet.next();
 
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);
diff --git a/flinkx-teradata/flinkx-teradata-reader/src/main/java/com/dtstack/flinkx/teradata/format/TeradataInputFormat.java b/flinkx-teradata/flinkx-teradata-reader/src/main/java/com/dtstack/flinkx/teradata/format/TeradataInputFormat.java
@@ -80,7 +80,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {
             if(splitWithRowCol){
                 columnCount = columnCount-1;
             }
-
+            checkSize(columnCount, metaColumns);
             hasNext = resultSet.next();
 
             descColumnTypeList = DbUtil.analyzeColumnType(resultSet);

Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`72`	`72`	`if(splitWithRowCol){`
`73`	`73`	`columnCount = columnCount-1;`
`74`	`74`	`}`
`75`		`-`
	`75`	`+ checkSize(columnCount, metaColumns);`
`76`	`76`	`hasNext = resultSet.next();`
`77`	`77`
`78`	`78`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`68`	`68`	`if(splitWithRowCol){`
`69`	`69`	`columnCount = columnCount-1;`
`70`	`70`	`}`
`71`		`-`
	`71`	`+ checkSize(columnCount, metaColumns);`
`72`	`72`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`
`73`	`73`
`74`	`74`	`} catch (SQLException se) {`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`78`	`78`	`if(splitWithRowCol){`
`79`	`79`	`columnCount = columnCount-1;`
`80`	`80`	`}`
`81`		`-`
	`81`	`+ checkSize(columnCount, metaColumns);`
`82`	`82`	`hasNext = resultSet.next();`
`83`	`83`
`84`	`84`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`
Original file line number	Diff line number	Diff line change
`@@ -186,7 +186,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`186`	`186`	`if (splitWithRowCol) {`
`187`	`187`	`columnCount = columnCount - 1;`
`188`	`188`	`}`
`189`		`-`
	`189`	`+ checkSize(columnCount, metaColumns);`
`190`	`190`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`
`191`	`191`
`192`	`192`	`} catch (SQLException se) {`
`@@ -832,4 +832,17 @@ private void queryStartLocation() throws SQLException{`
`832`	`832`	`}`
`833`	`833`	`}`
`834`	`834`
	`835`	`+ /**`
	`836`	`+ * 校验columnCount和metaColumns的长度是否相等`
	`837`	`+ * @param columnCount`
	`838`	`+ * @param metaColumns`
	`839`	`+ */`
	`840`	`+ protected void checkSize(int columnCount, List<MetaColumn> metaColumns) {`
	`841`	`+ if (!ConstantValue.STAR_SYMBOL.equals(metaColumns.get(0).getName()) && columnCount != metaColumns.size()) {`
	`842`	`+ LOG.error("error config: column size for reader is {},but columns size for query result is {}." +`
	`843`	`+ " And the query sql is '{}'.",`
	`844`	`+ metaColumns.size(), columnCount, querySql);`
	`845`	`+ throw new RuntimeException("");`
	`846`	`+ }`
	`847`	`+ }`
`835`	`848`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`76`	`76`	`if(splitWithRowCol){`
`77`	`77`	`columnCount = columnCount-1;`
`78`	`78`	`}`
`79`		`-`
	`79`	`+ checkSize(columnCount, metaColumns);`
`80`	`80`	`hasNext = resultSet.next();`
`81`	`81`
`82`	`82`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ public void openInternal(InputSplit inputSplit) throws IOException {`
`80`	`80`	`if(splitWithRowCol){`
`81`	`81`	`columnCount = columnCount-1;`
`82`	`82`	`}`
`83`		`-`
	`83`	`+ checkSize(columnCount, metaColumns);`
`84`	`84`	`hasNext = resultSet.next();`
`85`	`85`
`86`	`86`	`descColumnTypeList = DbUtil.analyzeColumnType(resultSet);`