[
https://issues.apache.org/jira/browse/KYLIN-3707?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16722153#comment-16722153
]
ASF GitHub Bot commented on KYLIN-3707:
---------------------------------------
shaofengshi closed pull request #366: KYLIN-3707 add configuration for setting
isolation-level for sqoop
URL: https://github.com/apache/kylin/pull/366
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/JdbcConnector.java
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/JdbcConnector.java
index d849e6c010..47ba6b3bc0 100644
---
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/JdbcConnector.java
+++
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/JdbcConnector.java
@@ -37,7 +37,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
public class JdbcConnector implements Closeable {
@@ -175,8 +174,7 @@ public String getPropertyValue(String key) {
return jdbcDs.getPropertyValue(key);
}
- @VisibleForTesting
- SqlConverter getSqlConverter() {
+ public SqlConverter getSqlConverter() {
return sqlConverter;
}
diff --git
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/DefaultConfiguer.java
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/DefaultConfiguer.java
index 6d7fb6da37..6c01a70072 100644
---
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/DefaultConfiguer.java
+++
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/DefaultConfiguer.java
@@ -78,10 +78,6 @@ public String fixAfterDefaultConvert(String orig) {
if (this.adaptor == null) {
return orig;
}
- // fix problem of case sensitive when generate sql.
-// if (isCaseSensitive()) {
-// orig = adaptor.fixCaseSensitiveSql(orig);
-// }
return adaptor.fixSql(orig);
}
@@ -134,4 +130,9 @@ public String fixIdentifierCaseSensitve(String orig) {
}
return adaptor.fixIdentifierCaseSensitve(orig);
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return dsDef.getPropertyValue("transaction.isolation-level");
+ }
}
diff --git
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverter.java
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverter.java
index d25c04fd61..e8302e8a23 100644
---
a/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverter.java
+++
b/datasource-sdk/src/main/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverter.java
@@ -110,5 +110,11 @@ public IConfigurer getConfigurer() {
boolean enableQuote();
String fixIdentifierCaseSensitve(String orig);
+
+ /**
+ * Only support following 3 types
+ *
TRANSACTION_READ_COMMITTED，TRANSACTION_READ_UNCOMMITTED，TRANSACTION_READ_COMMITTED
+ */
+ String getTransactionIsolationLevel();
}
}
diff --git
a/datasource-sdk/src/test/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverterTest.java
b/datasource-sdk/src/test/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverterTest.java
index 94cc651223..451be6061c 100644
---
a/datasource-sdk/src/test/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverterTest.java
+++
b/datasource-sdk/src/test/java/org/apache/kylin/sdk/datasource/framework/conv/SqlConverterTest.java
@@ -111,12 +111,20 @@ public boolean enableQuote() {
public String fixIdentifierCaseSensitve(String orig) {
return orig;
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return null;
+ }
}, master);
// escape default keywords
- Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select * from \"DEFAULT\".FACT"));
- Assert.assertEquals("SELECT *\nFROM \"Default\".\"FACT\"",
converter.convertSql("select * from \"Default\".FACT"));
- Assert.assertEquals("SELECT *\nFROM \"default\".\"FACT\"",
converter.convertSql("select * from \"default\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
+ converter.convertSql("select * from \"DEFAULT\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"Default\".\"FACT\"",
+ converter.convertSql("select * from \"Default\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"default\".\"FACT\"",
+ converter.convertSql("select * from \"default\".FACT"));
}
@Test
@@ -189,6 +197,11 @@ public boolean enableQuote() {
public String fixIdentifierCaseSensitve(String orig) {
return orig;
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return null;
+ }
}, master);
// normal cases
@@ -203,17 +216,20 @@ public String fixIdentifierCaseSensitve(String orig) {
// escape default keywords
Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select * from DEFAULT.FACT"));
- Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select * from \"DEFAULT\".FACT"));
- Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select * from \"Default\".FACT"));
- Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select * from \"default\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
+ converter.convertSql("select * from \"DEFAULT\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
+ converter.convertSql("select * from \"Default\".FACT"));
+ Assert.assertEquals("SELECT *\nFROM \"DEFAULT\".\"FACT\"",
+ converter.convertSql("select * from \"default\".FACT"));
// function mapping
Assert.assertEquals("SELECT EXTRACT(DOY FROM \"PART_DT\")\nFROM
\"DEFAULT\".\"FACT\"",
converter.convertSql("select DAYOFYEAR(PART_DT) from
\"DEFAULT\".FACT"));
Assert.assertEquals(
- "SELECT 12 * (EXTRACT(YEAR FROM \"DT1\") - EXTRACT(YEAR FROM
\"DT2\")) + EXTRACT(MONTH FROM \"DT1\") - EXTRACT(MONTH FROM \"DT2\") - " +
- "CASE WHEN EXTRACT(DAY FROM \"DT2\") > EXTRACT(DAY
FROM \"DT1\") THEN 1 ELSE 0 END\n" +
- "FROM \"DEFAULT\".\"FACT\"",
+ "SELECT 12 * (EXTRACT(YEAR FROM \"DT1\") - EXTRACT(YEAR FROM
\"DT2\")) + EXTRACT(MONTH FROM \"DT1\") - EXTRACT(MONTH FROM \"DT2\") - "
+ + "CASE WHEN EXTRACT(DAY FROM \"DT2\") > EXTRACT(DAY
FROM \"DT1\") THEN 1 ELSE 0 END\n"
+ + "FROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select TIMESTAMPDIFF(month,DT2,
DT1) from \"DEFAULT\".FACT"));
Assert.assertEquals("SELECT TRUNC(\"ID\")\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql("select cast(ID as INT) from
\"DEFAULT\".FACT"));
@@ -221,12 +237,14 @@ public String fixIdentifierCaseSensitve(String orig) {
converter.convertSql("select 1 from a where 1 BETWEEN 0 and
2"));
Assert.assertEquals("SELECT \"CURRENT_DATE\", TEST_CURR_TIME()",
converter.convertSql("select CURRENT_DATE, CURRENT_TIME"));
- Assert.assertEquals("SELECT EXP(AVG(LN(EXTRACT(DOY FROM
CAST('2018-03-20' AS DATE)))))\nFROM \"DEFAULT\".\"FACT\"",
+ Assert.assertEquals(
+ "SELECT EXP(AVG(LN(EXTRACT(DOY FROM CAST('2018-03-20' AS
DATE)))))\nFROM \"DEFAULT\".\"FACT\"",
converter.convertSql(
"select exp(avg(ln(dayofyear(cast('2018-03-20' as
date))))) from \"DEFAULT\".FACT"));
// over function
- Assert.assertEquals("SELECT STDDEVP(\"C1\") OVER (ORDER BY
\"C1\")\nFROM \"TEST_SUITE\"\nFETCH NEXT 1 ROWS ONLY",
+ Assert.assertEquals(
+ "SELECT STDDEVP(\"C1\") OVER (ORDER BY \"C1\")\nFROM
\"TEST_SUITE\"\nFETCH NEXT 1 ROWS ONLY",
converter.convertSql("select stddev_pop(c1) over(order by c1)
from test_suite limit 1"));
// type mapping
@@ -332,6 +350,11 @@ public boolean enableQuote() {
public String fixIdentifierCaseSensitve(String orig) {
return orig;
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return null;
+ }
}, master);
Assert.assertEquals("SELECT 1\nORDER BY 2\nOFFSET 0 ROWS\nFETCH NEXT 1
ROWS ONLY",
@@ -347,6 +370,7 @@ public String fixIdentifierCaseSensitve(String orig) {
Assert.assertEquals("SELECT 1\nORDER BY 1\nOFFSET 0 ROWS\nFETCH NEXT 1
ROWS ONLY",
converter.convertSql("SELECT 1 LIMIT 1"));
}
+
@Test
public void testConvertQuotedSqlWithEscape() throws SQLException {
DataSourceDefProvider provider = DataSourceDefProvider.getInstance();
@@ -417,6 +441,11 @@ public boolean enableQuote() {
public String fixIdentifierCaseSensitve(String orig) {
return orig;
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return null;
+ }
}, master);
Assert.assertEquals("SELECT SUM(\"A\"), COUNT(\"A\") AS \"AB\"\nFROM
\"DEFAULT\".\"CUBE\"",
@@ -425,8 +454,10 @@ public String fixIdentifierCaseSensitve(String orig) {
converter.convertSql("select A(), B(`A`), cast(`PRICE@@` as
`DDD`) from DEFAULT.`CUBE`"));
Assert.assertEquals("SELECT A(), B(\"A\"), CAST(\"PRICE@@\" AS
DDD)\nFROM \"DEFAULT\".\"CUBE\"",
converter.convertSql("select A(), B(\"A\"), cast(\"PRICE@@\"
as \"DDD\") from \"DEFAULT\".\"CUBE\""));
- Assert.assertEquals("SELECT \"kylin_sales\".\"price_@@\",
\"kylin_sales\".\"count\"\nFROM \"cube\".\"kylin_sales\"\nWHERE
\"kylin_sales\".\"price_@@\" > 1 AND \"kylin_sales\".\"count\" < 50",
- converter.convertSql("select `kylin_sales`.`price_@@`,
`kylin_sales`.`count` from `cube`.`kylin_sales` where `kylin_sales`.`price_@@`
> 1 and `kylin_sales`.`count` < 50"));
+ Assert.assertEquals(
+ "SELECT \"kylin_sales\".\"price_@@\",
\"kylin_sales\".\"count\"\nFROM \"cube\".\"kylin_sales\"\nWHERE
\"kylin_sales\".\"price_@@\" > 1 AND \"kylin_sales\".\"count\" < 50",
+ converter.convertSql(
+ "select `kylin_sales`.`price_@@`,
`kylin_sales`.`count` from `cube`.`kylin_sales` where `kylin_sales`.`price_@@`
> 1 and `kylin_sales`.`count` < 50"));
Assert.assertEquals("SELECT COUNT(DISTINCT \"price_#@\")\nFROM
\"cube\".\"kylin_sales\"",
converter.convertSql("select count(distinct `price_#@`) from
`cube`.`kylin_sales`"));
@@ -502,6 +533,11 @@ public boolean enableQuote() {
public String fixIdentifierCaseSensitve(String orig) {
return orig.toUpperCase(Locale.ROOT);
}
+
+ @Override
+ public String getTransactionIsolationLevel() {
+ return null;
+ }
}, master);
Assert.assertEquals("\"TEST\".\"AA\"",
converter.convertColumn("`test`.`aa`", "`"));
diff --git
a/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/extensible/JdbcHiveInputBase.java
b/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/extensible/JdbcHiveInputBase.java
index 10eb31ee21..ec6908451a 100644
---
a/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/extensible/JdbcHiveInputBase.java
+++
b/source-jdbc/src/main/java/org/apache/kylin/source/jdbc/extensible/JdbcHiveInputBase.java
@@ -29,6 +29,7 @@
import org.apache.kylin.metadata.model.SegmentRange;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.sdk.datasource.framework.JdbcConnector;
+import org.apache.kylin.sdk.datasource.framework.conv.SqlConverter;
import org.apache.kylin.source.jdbc.sqoop.SqoopCmdStep;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,15 +83,15 @@ protected AbstractExecutable
createSqoopToFlatHiveStep(String jobWorkingDir, Str
String filedDelimiter = config.getJdbcSourceFieldDelimiter();
int mapperNum = config.getSqoopMapperNum();
- String bquery = String.format(Locale.ROOT, "SELECT min(%s),
max(%s) FROM `%s`.%s as `%s`", splitColumn, splitColumn,
- splitDatabase, splitTable, splitTableAlias);
+ String bquery = String.format(Locale.ROOT, "SELECT min(%s),
max(%s) FROM `%s`.%s as `%s`", splitColumn,
+ splitColumn, splitDatabase, splitTable, splitTableAlias);
bquery = dataSource.convertSql(bquery);
if (partitionDesc.isPartitioned()) {
SegmentRange segRange = flatDesc.getSegRange();
if (segRange != null && !segRange.isInfinite()) {
if
(partitionDesc.getPartitionDateColumnRef().getTableAlias().equals(splitTableAlias)
&& (partitionDesc.getPartitionTimeColumnRef() ==
null || partitionDesc
-
.getPartitionTimeColumnRef().getTableAlias().equals(splitTableAlias))) {
+
.getPartitionTimeColumnRef().getTableAlias().equals(splitTableAlias))) {
String quotedPartCond =
FlatTableSqlQuoteUtils.quoteIdentifierInSqlExpr(flatDesc,
partitionDesc.getPartitionConditionBuilder().buildDateRangeCondition(partitionDesc,
flatDesc.getSegment(), segRange),
@@ -110,6 +111,11 @@ protected AbstractExecutable
createSqoopToFlatHiveStep(String jobWorkingDir, Str
dataSource.getJdbcUrl(), dataSource.getJdbcDriver(),
dataSource.getJdbcUser(),
dataSource.getJdbcPassword(), selectSql, jobWorkingDir,
hiveTable, splitColumn, bquery,
filedDelimiter, mapperNum);
+ SqlConverter.IConfigurer configurer =
dataSource.getSqlConverter().getConfigurer();
+ if (configurer.getTransactionIsolationLevel() != null) {
+ cmd = cmd + " --relaxed-isolation
--metadata-transaction-isolation-level "
+ + configurer.getTransactionIsolationLevel();
+ }
logger.debug("sqoop cmd: {}", cmd);
SqoopCmdStep step = new SqoopCmdStep();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Add configuration for setting isolation-level for sqoop
> -------------------------------------------------------
>
> Key: KYLIN-3707
> URL: https://issues.apache.org/jira/browse/KYLIN-3707
> Project: Kylin
> Issue Type: Improvement
> Components: RDBMS Source
> Environment: MacOSX JDK1.8
> Reporter: rongchuan.jin
> Priority: Minor
>
> When use RDBMS as datasource,we use Apache Sqoop to import data from RDBMS to
> HDFS.Sqoop use read_commited isolation-level by default,while some RDBMS does
> not
> support read_commited level,so need to add a configuration to fix it.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
