[Hotfix][Connector-V2][Hive] Fix the bug that when write data to hive…

… throws NullPointerException
apache · Nov 1, 2022 · 907ecad · 907ecad
1 parent 9bd076c
commit 907ecad
Show file tree

Hide file tree

Showing 5 changed files with 15 additions and 5 deletions.
diff --git a/...n/java/org/apache/seatunnel/connectors/seatunnel/file/sink/config/TextFileSinkConfig.java b/...n/java/org/apache/seatunnel/connectors/seatunnel/file/sink/config/TextFileSinkConfig.java
@@ -130,12 +130,12 @@ public TextFileSinkConfig(@NonNull Config config, @NonNull SeaTunnelRowType seaT
         Map<String, Integer> columnsMap = new HashMap<>(seaTunnelRowTypeInfo.getFieldNames().length);
         String[] fieldNames = seaTunnelRowTypeInfo.getFieldNames();
         for (int i = 0; i < fieldNames.length; i++) {
-            columnsMap.put(fieldNames[i], i);
+            columnsMap.put(fieldNames[i].toLowerCase(), i);
         }
 
         // init sink column index and partition field index, we will use the column index to found the data in SeaTunnelRow
         this.sinkColumnsIndexInRow = this.sinkColumnList.stream()
-            .map(columnsMap::get)
+            .map(column -> columnsMap.get(column.toLowerCase()))
             .collect(Collectors.toList());
 
         if (!CollectionUtils.isEmpty(this.partitionFieldList)) {

diff --git a/...ava/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java b/...ava/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/AbstractWriteStrategy.java
@@ -248,7 +248,7 @@ public void beginTransaction(Long checkpointId) {
      * @return transaction ids
      */
     public List<String> getTransactionIdFromStates(List<FileSinkState> fileStates) {
-        String[] pathSegments = new String[]{textFileSinkConfig.getPath(), Constant.SEATUNNEL, jobId};
+        String[] pathSegments = new String[]{textFileSinkConfig.getTmpPath(), Constant.SEATUNNEL, jobId};
         String jobDir = String.join(File.separator, pathSegments) + File.separator;
         try {
             List<String> transactionDirList = FileSystemUtils.dirList(jobDir).stream().map(Path::toString).collect(Collectors.toList());

diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/OrcWriteStrategy.java
@@ -179,7 +179,7 @@ private TypeDescription buildSchemaWithRowType() {
         TypeDescription schema = TypeDescription.createStruct();
         for (Integer i : sinkColumnsIndexInRow) {
             TypeDescription fieldType = buildFieldWithRowType(seaTunnelRowType.getFieldType(i));
-            schema.addField(seaTunnelRowType.getFieldName(i), fieldType);
+            schema.addField(seaTunnelRowType.getFieldName(i).toLowerCase(), fieldType);
         }
         return schema;
     }

diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java b/...java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ParquetWriteStrategy.java
@@ -95,7 +95,7 @@ public void write(@NonNull SeaTunnelRow seaTunnelRow) {
         for (Integer integer : sinkColumnsIndexInRow) {
             String fieldName = seaTunnelRowType.getFieldName(integer);
             Object field = seaTunnelRow.getField(integer);
-            recordBuilder.set(fieldName, resolveObject(field, seaTunnelRowType.getFieldType(integer)));
+            recordBuilder.set(fieldName.toLowerCase(), resolveObject(field, seaTunnelRowType.getFieldType(integer)));
         }
         GenericData.Record record = recordBuilder.build();
         try {

diff --git a/seatunnel-connectors-v2/connector-hive/pom.xml b/seatunnel-connectors-v2/connector-hive/pom.xml
@@ -50,6 +50,12 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-shaded-hadoop-2</artifactId>
             <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <artifactId>avro</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.apache.hive</groupId>
@@ -89,6 +95,10 @@
                     <artifactId>pentaho-aggdesigner-algorithm</artifactId>
                     <groupId>org.pentaho</groupId>
                 </exclusion>
+                <exclusion>
+                    <artifactId>avro</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                </exclusion>
             </exclusions>
         </dependency>
     </dependencies>