private void executeBatchSql(String tempSql,
                                 String schema,
                                 String tableName,
                                 String storeType,
                                 Boolean enablePartition,
                                 List<String> fieldNames,
                                 String partitionFields,
                                 Map<String, ArrayList<String>> rowDataMap) {
        StringBuilder partitionCondition = new StringBuilder();
        String tableFieldsCondition = buildTableFieldsCondition(fieldNames, partitionFields);
        ArrayList<String> rowData = new ArrayList<>();
        String tableNameInfo = Objects.isNull(schema) ?
                tableName : quoteIdentifier(schema) + "." + tableName;
        tempSql = tempSql.replace("tableName", tableNameInfo);
        boolean isPartitioned = storeType.equalsIgnoreCase(KUDU_TYPE) || !enablePartition;

        try {
            // kudu ${partitionCondition} is null
            if (isPartitioned) {
                tempSql = tempSql
                    .replace(PARTITION_CONDITION, partitionCondition.toString())
                    .replace(PARTITION_CONSTANT, "")
                    .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition);
                rowData.addAll(rowDataMap.get(NO_PARTITION));
                String executeSql = tempSql.replace(VALUES_CONDITION, String.join(", ", rowData));
                statement.execute(executeSql);
                rowData.clear();
            } else {
                // partition sql
                Set<String> keySet = rowDataMap.keySet();
                for (String key : keySet) {
                    rowData.addAll(rowDataMap.get(key));
                    partitionCondition.append(key);
                    tempSql = tempSql
                        .replace(PARTITION_CONDITION, partitionCondition.toString())
                        .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition);
                    String executeSql = tempSql
                        .replace(VALUES_CONDITION, String.join(", ", rowData));
                    statement.execute(executeSql);
                    partitionCondition.delete(0, partitionCondition.length());
                }
            }
        } catch (Exception e) {
            if (e instanceof SQLException) {
                dealBatchSqlError(rowData, connection, statement, tempSql);
            } else {
                throw new RuntimeException("Insert into impala error!", e);
            }
        } finally {
            rowData.clear();
        }
    }
execute batch sql from row data map
sql like 'insert into tableName(f1, f2, f3) ${partitionCondition} values(v1, v2, v3), (v4, v5, v6)....

@param tempSql         template sql
@param storeType       the store type of data
@param enablePartition enable partition or not
@param fieldNames      field name list
@param partitionFields partition fields
@param rowDataMap      row data map
impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java