Alawn/Druid
1
0
Fork 0
mirror of https://github.com/alibaba/druid.git synced 2023-03-04 19:17:16 +00:00
Code Issues Releases Wiki Activity

improved odps support, fix #3884

Browse Source
This commit is contained in:
高铁
2020-10-05 22:59:01 +08:00
parent 0fa21ab918
commit ad8d37cd4a
6 changed files with 342 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/alibaba/druid/sql/dialect/hive/parser/HiveStatementParser.java
+35 -30
View File
@@ -150,36 +150,7 @@ public class HiveStatementParser extends SQLStatementParser {
}
if (lexer.identifierEquals(FnvHash.Constants.LOAD)) {
lexer.nextToken();
acceptIdentifier("DATA");
HiveLoadDataStatement stmt = new HiveLoadDataStatement();
if (lexer.identifierEquals(FnvHash.Constants.LOCAL)) {
lexer.nextToken();
stmt.setLocal(true);
}
acceptIdentifier("INPATH");
SQLExpr inpath = this.exprParser.expr();
stmt.setInpath(inpath);
if (lexer.token() == Token.OVERWRITE) {
lexer.nextToken();
stmt.setOverwrite(true);
}
accept(Token.INTO);
accept(Token.TABLE);
SQLExpr table = this.exprParser.expr();
stmt.setInto(table);
if (lexer.token() == Token.PARTITION) {
lexer.nextToken();
accept(Token.LPAREN);
this.exprParser.exprList(stmt.getPartition(), stmt);
accept(Token.RPAREN);
}
HiveLoadDataStatement stmt = parseLoad();
statementList.add(stmt);
return true;
@@ -345,6 +316,40 @@ public class HiveStatementParser extends SQLStatementParser {
return false;
}
protected HiveLoadDataStatement parseLoad() {
acceptIdentifier("LOAD");
acceptIdentifier("DATA");
HiveLoadDataStatement stmt = new HiveLoadDataStatement();
if (lexer.identifierEquals(FnvHash.Constants.LOCAL)) {
lexer.nextToken();
stmt.setLocal(true);
}
acceptIdentifier("INPATH");
SQLExpr inpath = this.exprParser.expr();
stmt.setInpath(inpath);
if (lexer.token() == Token.OVERWRITE) {
lexer.nextToken();
stmt.setOverwrite(true);
}
accept(Token.INTO);
accept(Token.TABLE);
SQLExpr table = this.exprParser.expr();
stmt.setInto(table);
if (lexer.token() == Token.PARTITION) {
lexer.nextToken();
accept(Token.LPAREN);
this.exprParser.exprList(stmt.getPartition(), stmt);
accept(Token.RPAREN);
}
return stmt;
}
public SQLCreateTableStatement parseCreateTable() {
SQLCreateTableParser parser = new HiveCreateTableParser(this.exprParser);
return parser.parseCreateTable();
src/main/java/com/alibaba/druid/sql/dialect/hive/stmt/HiveLoadDataStatement.java
+61
View File
@@ -6,11 +6,14 @@ import com.alibaba.druid.sql.ast.SQLObject;
import com.alibaba.druid.sql.ast.SQLStatementImpl;
import com.alibaba.druid.sql.ast.statement.SQLAssignItem;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.ast.statement.SQLExternalRecordFormat;
import com.alibaba.druid.sql.dialect.hive.visitor.HiveASTVisitor;
import com.alibaba.druid.sql.visitor.SQLASTVisitor;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
public class HiveLoadDataStatement extends SQLStatementImpl {
private boolean local;
@@ -20,6 +23,12 @@ public class HiveLoadDataStatement extends SQLStatementImpl {
private final List<SQLExpr> partition = new ArrayList<SQLExpr>(4);
private SQLExternalRecordFormat format;
private SQLExpr storedBy;
private SQLExpr storedAs;
private SQLExpr rowFormat;
protected Map<String, SQLObject> serdeProperties = new LinkedHashMap<String, SQLObject>();
public HiveLoadDataStatement() {
super(DbType.hive);
}
@@ -37,6 +46,9 @@ public class HiveLoadDataStatement extends SQLStatementImpl {
this.acceptChild(visitor, inpath);
this.acceptChild(visitor, into);
this.acceptChild(visitor, partition);
this.acceptChild(visitor, storedAs);
this.acceptChild(visitor, storedBy);
this.acceptChild(visitor, rowFormat);
}
visitor.endVisit(this);
}
@@ -98,4 +110,53 @@ public class HiveLoadDataStatement extends SQLStatementImpl {
}
this.partition.add(item);
}
public SQLExternalRecordFormat getFormat() {
return format;
}
public void setFormat(SQLExternalRecordFormat x) {
if (x != null) {
x.setParent(this);
}
this.format = x;
}
public SQLExpr getStoredBy() {
return storedBy;
}
public void setStoredBy(SQLExpr x) {
if (x != null) {
x.setParent(this);
}
this.storedBy = x;
}
public SQLExpr getStoredAs() {
return storedAs;
}
public void setStoredAs(SQLExpr x) {
if (x != null) {
x.setParent(this);
}
this.storedAs = x;
}
public SQLExpr getRowFormat() {
return rowFormat;
}
public void setRowFormat(SQLExpr x) {
if (x != null) {
x.setParent(this);
}
this.rowFormat = x;
}
public Map<String, SQLObject> getSerdeProperties() {
return serdeProperties;
}
}
src/main/java/com/alibaba/druid/sql/dialect/odps/parser/OdpsStatementParser.java
+95 -1
View File
@@ -21,7 +21,7 @@ import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.statement.*;
import com.alibaba.druid.sql.dialect.hive.ast.HiveInsert;
import com.alibaba.druid.sql.dialect.hive.ast.HiveMultiInsertStatement;
import com.alibaba.druid.sql.dialect.hive.stmt.HiveMsckRepairStatement;
import com.alibaba.druid.sql.dialect.hive.stmt.HiveLoadDataStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlKillStatement;
import com.alibaba.druid.sql.dialect.odps.ast.*;
import com.alibaba.druid.sql.parser.*;
@@ -29,6 +29,8 @@ import com.alibaba.druid.util.FnvHash;
import java.util.List;
import static com.alibaba.druid.sql.parser.Token.OVERWRITE;
public class OdpsStatementParser extends SQLStatementParser {
public OdpsStatementParser(String sql){
@@ -294,6 +296,13 @@ public class OdpsStatementParser extends SQLStatementParser {
return true;
}
if (lexer.identifierEquals(FnvHash.Constants.LOAD)) {
HiveLoadDataStatement stmt = parseLoad();
statementList.add(stmt);
return true;
}
if (lexer.token() == Token.VARIANT && lexer.stringVal().startsWith("@")) {
Lexer.SavePoint mark = lexer.mark();
String variant = lexer.stringVal();
@@ -821,4 +830,89 @@ public class OdpsStatementParser extends SQLStatementParser {
public SQLCreateFunctionStatement parseCreateFunction() {
return parseHiveCreateFunction();
}
protected HiveLoadDataStatement parseLoad() {
acceptIdentifier("LOAD");
HiveLoadDataStatement stmt = new HiveLoadDataStatement();
if (lexer.token() == OVERWRITE) {
stmt.setOverwrite(true);
lexer.nextToken();
}
accept(Token.TABLE);
stmt.setInto(this.exprParser.expr());
if (lexer.token() == Token.PARTITION) {
lexer.nextToken();
accept(Token.LPAREN);
this.exprParser.exprList(stmt.getPartition(), stmt);
accept(Token.RPAREN);
}
if (lexer.identifierEquals(FnvHash.Constants.LOCAL)) {
lexer.nextToken();
stmt.setLocal(true);
}
accept(Token.FROM);
acceptIdentifier("LOCATION");
SQLExpr inpath = this.exprParser.expr();
stmt.setInpath(inpath);
if (lexer.identifierEquals("STORED")) {
lexer.nextToken();
if (lexer.token() == Token.BY) {
lexer.nextToken();
stmt.setStoredBy(this.exprParser.expr());
} else {
accept(Token.AS);
stmt.setStoredAs(this.exprParser.expr());
}
}
if (lexer.identifierEquals("ROW")) {
lexer.nextToken();
acceptIdentifier("FORMAT");
acceptIdentifier("SERDE");
stmt.setRowFormat(this.exprParser.expr());
}
if (lexer.token() == Token.WITH) {
lexer.nextToken();
acceptIdentifier("SERDEPROPERTIES");
accept(Token.LPAREN);
for (;;) {
String name = lexer.stringVal();
lexer.nextToken();
accept(Token.EQ);
SQLExpr value = this.exprParser.primary();
stmt.getSerdeProperties().put(name, value);
if (lexer.token() == Token.COMMA) {
lexer.nextToken();
continue;
}
break;
}
accept(Token.RPAREN);
}
if (lexer.identifierEquals("STORED")) {
lexer.nextToken();
accept(Token.AS);
stmt.setStoredAs(this.exprParser.expr());
}
return stmt;
}
}
src/main/java/com/alibaba/druid/sql/dialect/odps/visitor/OdpsOutputVisitor.java
+49
View File
@@ -21,6 +21,7 @@ import com.alibaba.druid.sql.ast.expr.SQLCharExpr;
import com.alibaba.druid.sql.ast.statement.*;
import com.alibaba.druid.sql.ast.statement.SQLJoinTableSource.JoinType;
import com.alibaba.druid.sql.dialect.hive.stmt.HiveCreateTableStatement;
import com.alibaba.druid.sql.dialect.hive.stmt.HiveLoadDataStatement;
import com.alibaba.druid.sql.dialect.hive.visitor.HiveOutputVisitor;
import com.alibaba.druid.sql.dialect.odps.ast.*;
@@ -1037,4 +1038,52 @@ public class OdpsOutputVisitor extends HiveOutputVisitor implements OdpsASTVisit
public void endVisit(OdpsExstoreStatement x) {
}
@Override
public boolean visit(HiveLoadDataStatement x) {
print0(ucase ? "LOAD " : "load ");
if (x.isOverwrite()) {
print0(ucase ? "OVERWRITE " : "overwrite ");
}
print0(ucase ? "INTO TABLE " : "into table ");
x.getInto().accept(this);
if (x.getPartition().size() > 0) {
print0(ucase ? " PARTITION (" : " partition (");
printAndAccept(x.getPartition(), ", ");
print(')');
}
println();
print0(ucase ? "LOCATION " : "location ");
x.getInpath().accept(this);
SQLExpr storedBy = x.getStoredBy();
if (storedBy != null) {
println();
print0(ucase ? "STORED BY " : "stored by ");
storedBy.accept(this);
}
SQLExpr rowFormat = x.getRowFormat();
if (rowFormat != null) {
println();
print0(ucase ? "ROW FORMAT SERDE " : "row format serde ");
rowFormat.accept(this);
}
printSerdeProperties(x.getSerdeProperties());
SQLExpr storedAs = x.getStoredAs();
if (storedAs != null) {
println();
print0(ucase ? "STORED AS " : "stored as ");
storedAs.accept(this);
}
return false;
}
}
src/main/java/com/alibaba/druid/sql/visitor/SQLASTOutputVisitor.java
+41 -35
View File
@@ -10329,41 +10329,7 @@ public class SQLASTOutputVisitor extends SQLASTVisitorAdapter implements Paramet
}
Map<String, SQLObject> serdeProperties = x.getSerdeProperties();
if (serdeProperties.size() > 0) {
println();
print0(ucase ? "WITH SERDEPROPERTIES (" : "with serdeproperties (");
incrementIndent();
println();
int i = 0;
for (Map.Entry<String, SQLObject> option : serdeProperties.entrySet()) {
if (i != 0) {
print(",");
println();
}
String key = option.getKey();
boolean unquote = false;
char c0;
if (key.length() > 0 && (c0 = key.charAt(0)) != '"' && c0 != '`' && c0 != '\'') {
unquote = true;
}
if (unquote) {
print('\'');
}
print0(key);
if (unquote) {
print('\'');
}
print0(" = ");
option.getValue().accept(this);
++i;
}
decrementIndent();
println();
print(')');
}
printSerdeProperties(serdeProperties);
List<SQLSelectOrderByItem> sortedBy = x.getSortedBy();
if (sortedBy.size() > 0) {
@@ -10433,6 +10399,46 @@ public class SQLASTOutputVisitor extends SQLASTVisitorAdapter implements Paramet
}
}
protected void printSerdeProperties(Map<String, SQLObject> serdeProperties) {
if (serdeProperties.isEmpty()) {
return;
}
println();
print0(ucase ? "WITH SERDEPROPERTIES (" : "with serdeproperties (");
incrementIndent();
println();
int i = 0;
for (Map.Entry<String, SQLObject> option : serdeProperties.entrySet()) {
if (i != 0) {
print(",");
println();
}
String key = option.getKey();
boolean unquote = false;
char c0;
if (key.length() > 0 && (c0 = key.charAt(0)) != '"' && c0 != '`' && c0 != '\'') {
unquote = true;
}
if (unquote) {
print('\'');
}
print0(key);
if (unquote) {
print('\'');
}
print0(" = ");
option.getValue().accept(this);
++i;
}
decrementIndent();
println();
print(')');
}
protected void printTblProperties(HiveCreateTableStatement x) {
List<SQLAssignItem> tblProperties = x.getTblProperties();
if (tblProperties.size() > 0) {
src/test/java/com/alibaba/druid/bvt/sql/odps/OdpsLoadTest.java
+61
View File
@@ -0,0 +1,61 @@
package com.alibaba.druid.bvt.sql.odps;
import com.alibaba.druid.sql.SQLUtils;
import junit.framework.TestCase;
public class OdpsLoadTest extends TestCase {
public void test_load_0() throws Exception {
String sql = "LOAD OVERWRITE TABLE tsv_load_tbl \n" +
"FROM\n" +
"LOCATION 'oss://accessId:accesssKey@oss-cn-hangzhou-zmf.aliyuncs.com/my_bucket_id/my_location/'\n" +
"STORED BY 'com.aliyun.odps.TsvStorageHandler'\n" +
"WITH SERDEPROPERTIES ('odps.text.option.delimiter'='\\t');";
assertEquals("LOAD OVERWRITE INTO TABLE tsv_load_tbl\n" +
"LOCATION 'oss://accessId:accesssKey@oss-cn-hangzhou-zmf.aliyuncs.com/my_bucket_id/my_location/'\n" +
"STORED BY 'com.aliyun.odps.TsvStorageHandler'\n" +
"WITH SERDEPROPERTIES (\n" +
"\t'odps.text.option.delimiter' = '\t'\n" +
");", SQLUtils.formatOdps(sql));
}
public void test_load_1() throws Exception {
String sql = "LOAD OVERWRITE TABLE oss_load_static_part PARTITION(ds='20190101')\n" +
"FROM\n" +
"LOCATION 'oss://<yourAccessKeyId>:<yourAccessKeySecret>@oss-cn-hangzhou-zmf.aliyuncs.com/my_bucket_id/my_location/'\n" +
"STORED AS PARQUET;";
assertEquals("LOAD OVERWRITE INTO TABLE oss_load_static_part PARTITION (ds = '20190101')\n" +
"LOCATION 'oss://<yourAccessKeyId>:<yourAccessKeySecret>@oss-cn-hangzhou-zmf.aliyuncs.com/my_bucket_id/my_location/'\n" +
"STORED AS PARQUET;", SQLUtils.formatOdps(sql));
}
public void test_load_2() throws Exception {
String sql = "LOAD OVERWRITE TABLE oss_load_dyn_part PARTITION(ds)\n" +
"FROM\n" +
"LOCATION 'oss://accessId:accesssKey@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/'\n" +
"ROW FORMAT serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'\n" +
"WITH SERDEPROPERTIES(\n" +
" 'Fields terminator'='\\001',\n" +
" 'Escape delimitor'='\\\\',\n" +
" 'Collection items terminator'='\\002',\n" +
" 'Map keys terminator'='\\003',\n" +
" 'Lines terminator'='\\n',\n" +
" 'Null defination'='\\\\N')\n" +
"STORED AS TEXTFILE ;";
assertEquals("LOAD OVERWRITE INTO TABLE oss_load_dyn_part PARTITION (ds)\n" +
"LOCATION 'oss://accessId:accesssKey@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/'\n" +
"ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'\n" +
"WITH SERDEPROPERTIES (\n" +
"\t'Fields terminator' = '\\001',\n" +
"\t'Escape delimitor' = '\\\\',\n" +
"\t'Collection items terminator' = '\\002',\n" +
"\t'Map keys terminator' = '\\003',\n" +
"\t'Lines terminator' = '\\n',\n" +
"\t'Null defination' = '\\\\N'\n" +
")\n" +
"STORED AS TEXTFILE;", SQLUtils.formatOdps(sql));
}
}