-
Notifications
You must be signed in to change notification settings - Fork 1.8k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[Feature][Connector-V2][Jdbc] support gbase 8a #3026
Merged
Merged
Changes from 21 commits
Commits
Show all changes
27 commits
Select commit
Hold shift + click to select a range
eb44fc4
gbase 8a connector
liugddx e8a1ecc
Merge remote-tracking branch 'upstream/dev' into jdbc-gbase8a
liugddx 74928f9
add gbase8a e2e test
liugddx 9971e65
add gbase8a e2e test
liugddx c1c86a7
fix some error
liugddx 6582058
fix some error
liugddx f524864
FIX time type error
liugddx 106db76
FIX time type error
liugddx 186418b
FIX time type error
liugddx cc26aa8
FIX time type error
liugddx 6f979b6
add gbase8a doc
liugddx 1df87b6
Merge remote-tracking branch 'upstream/dev' into jdbc-gbase8a
liugddx f7845c4
reslove merge error
liugddx f13c34f
fix oracle connector error
liugddx 373e8f9
fix oracle connector error
liugddx d4fb795
fix oracle float precision problem
liugddx e13d2d5
Merge remote-tracking branch 'upstream/dev' into jdbc-gbase8a
liugddx 271e968
fix some error
liugddx ec75d10
fix some error
liugddx 51f05c6
Merge remote-tracking branch 'upstream/dev' into jdbc-gbase8a
liugddx b14234c
Update seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apach…
liugddx d0430dd
Merge branch 'dev' into jdbc-gbase8a
Hisoka-X 52a0de4
fix jdbc e2e test error
liugddx 179aa2f
revert gbase8a date type.
liugddx 84eef7e
modify gbase8a e2e test
liugddx fe7f505
fix jdbc connector e2e bug
liugddx b3b1779
revert code
liugddx File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 71 additions & 0 deletions
71
seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SqlDateType.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,71 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.seatunnel.api.table.type; | ||
|
||
import java.sql.Date; | ||
import java.sql.Time; | ||
import java.sql.Timestamp; | ||
import java.util.Objects; | ||
|
||
public class SqlDateType<T> implements SeaTunnelDataType<T> { | ||
private static final long serialVersionUID = 2L; | ||
|
||
public static final SqlDateType<Date> SQL_DATE_TYPE = new SqlDateType<>(Date.class, SqlType.DATE); | ||
public static final SqlDateType<Time> SQL_TIME_TYPE = new SqlDateType<>(Time.class, SqlType.TIME); | ||
public static final SqlDateType<Timestamp> SQL_DATE_TIME_TYPE = new SqlDateType<>(Timestamp.class, SqlType.TIMESTAMP); | ||
|
||
private final Class<T> typeClass; | ||
private final SqlType sqlType; | ||
|
||
private SqlDateType(Class<T> typeClass, SqlType sqlType) { | ||
this.typeClass = typeClass; | ||
this.sqlType = sqlType; | ||
} | ||
|
||
@Override | ||
public Class<T> getTypeClass() { | ||
return typeClass; | ||
} | ||
|
||
@Override | ||
public SqlType getSqlType() { | ||
return this.sqlType; | ||
} | ||
|
||
@Override | ||
public int hashCode() { | ||
return Objects.hash(typeClass); | ||
} | ||
|
||
@Override | ||
public boolean equals(Object obj) { | ||
if (obj == this) { | ||
return true; | ||
} | ||
if (!(obj instanceof SqlDateType)) { | ||
return false; | ||
} | ||
SqlDateType<?> that = (SqlDateType<?>) obj; | ||
return Objects.equals(typeClass, that.typeClass); | ||
} | ||
|
||
@Override | ||
public String toString() { | ||
return sqlType.toString(); | ||
} | ||
} |
39 changes: 39 additions & 0 deletions
39
...g/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/gbase8a/Gbase8aDialect.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,39 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a; | ||
|
||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter; | ||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect; | ||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper; | ||
|
||
public class Gbase8aDialect implements JdbcDialect { | ||
@Override | ||
public String dialectName() { | ||
return "Gbase8a"; | ||
} | ||
|
||
@Override | ||
public JdbcRowConverter getRowConverter() { | ||
return new Gbase8aJdbcRowConverter(); | ||
} | ||
|
||
@Override | ||
public JdbcDialectTypeMapper getJdbcDialectTypeMapper() { | ||
return new Gbase8aTypeMapper(); | ||
} | ||
} |
36 changes: 36 additions & 0 deletions
36
...e/seatunnel/connectors/seatunnel/jdbc/internal/dialect/gbase8a/Gbase8aDialectFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,36 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a; | ||
|
||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect; | ||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory; | ||
|
||
import com.google.auto.service.AutoService; | ||
|
||
@AutoService(JdbcDialectFactory.class) | ||
public class Gbase8aDialectFactory implements JdbcDialectFactory { | ||
@Override | ||
public boolean acceptsURL(String url) { | ||
return url.startsWith("jdbc:gbase:"); | ||
} | ||
|
||
@Override | ||
public JdbcDialect create() { | ||
return new Gbase8aDialect(); | ||
} | ||
} |
90 changes: 90 additions & 0 deletions
90
...seatunnel/connectors/seatunnel/jdbc/internal/dialect/gbase8a/Gbase8aJdbcRowConverter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,90 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a; | ||
|
||
import org.apache.seatunnel.api.table.type.SeaTunnelDataType; | ||
import org.apache.seatunnel.api.table.type.SeaTunnelRow; | ||
import org.apache.seatunnel.api.table.type.SeaTunnelRowType; | ||
import org.apache.seatunnel.api.table.type.SqlType; | ||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter; | ||
|
||
import java.math.BigDecimal; | ||
import java.math.BigInteger; | ||
import java.sql.ResultSet; | ||
import java.sql.ResultSetMetaData; | ||
import java.sql.SQLException; | ||
import java.util.ArrayList; | ||
import java.util.List; | ||
|
||
public class Gbase8aJdbcRowConverter extends AbstractJdbcRowConverter { | ||
@Override | ||
public String converterName() { | ||
return "Gbase8a"; | ||
} | ||
|
||
@SuppressWarnings("checkstyle:MagicNumber") | ||
@Override | ||
public SeaTunnelRow toInternal(ResultSet rs, ResultSetMetaData metaData, SeaTunnelRowType typeInfo) throws SQLException { | ||
List<Object> fields = new ArrayList<>(typeInfo.getFieldTypes().length); | ||
SeaTunnelDataType<?>[] seaTunnelDataTypes = typeInfo.getFieldTypes(); | ||
|
||
for (int i = 1; i <= seaTunnelDataTypes.length; i++) { | ||
Object seatunnelField; | ||
SqlType sqlType = seaTunnelDataTypes[i - 1].getSqlType(); | ||
if (null == rs.getObject(i)) { | ||
seatunnelField = null; | ||
} else if (SqlType.BOOLEAN.equals(sqlType)) { | ||
seatunnelField = rs.getBoolean(i); | ||
} else if (SqlType.TINYINT.equals(sqlType)) { | ||
seatunnelField = rs.getByte(i); | ||
} else if (SqlType.SMALLINT.equals(sqlType)) { | ||
seatunnelField = rs.getShort(i); | ||
} else if (SqlType.INT.equals(sqlType)) { | ||
seatunnelField = rs.getInt(i); | ||
} else if (SqlType.BIGINT.equals(sqlType)) { | ||
seatunnelField = rs.getLong(i); | ||
} else if (SqlType.DECIMAL.equals(sqlType)) { | ||
Object value = rs.getObject(i); | ||
seatunnelField = value instanceof BigInteger ? | ||
new BigDecimal((BigInteger) value, 0) | ||
: value; | ||
} else if (SqlType.FLOAT.equals(sqlType)) { | ||
seatunnelField = rs.getFloat(i); | ||
} else if (SqlType.DOUBLE.equals(sqlType)) { | ||
seatunnelField = rs.getDouble(i); | ||
} else if (SqlType.STRING.equals(sqlType)) { | ||
seatunnelField = rs.getString(i); | ||
} else if (SqlType.TIME.equals(sqlType)) { | ||
seatunnelField = rs.getTime(i); | ||
} else if (SqlType.DATE.equals(sqlType)) { | ||
seatunnelField = rs.getDate(i); | ||
} else if (SqlType.TIMESTAMP.equals(sqlType)) { | ||
seatunnelField = rs.getTimestamp(i); | ||
} else if (SqlType.BYTES.equals(sqlType)) { | ||
seatunnelField = rs.getBytes(i); | ||
} else { | ||
throw new IllegalStateException("Unexpected value: " + sqlType); | ||
} | ||
|
||
fields.add(seatunnelField); | ||
} | ||
|
||
return new SeaTunnelRow(fields.toArray()); | ||
} | ||
|
||
} |
110 changes: 110 additions & 0 deletions
110
...pache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/gbase8a/Gbase8aTypeMapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,110 @@ | ||
/* | ||
* Licensed to the Apache Software Foundation (ASF) under one or more | ||
* contributor license agreements. See the NOTICE file distributed with | ||
* this work for additional information regarding copyright ownership. | ||
* The ASF licenses this file to You under the Apache License, Version 2.0 | ||
* (the "License"); you may not use this file except in compliance with | ||
* the License. You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
|
||
package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.gbase8a; | ||
|
||
import org.apache.seatunnel.api.table.type.BasicType; | ||
import org.apache.seatunnel.api.table.type.DecimalType; | ||
import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; | ||
import org.apache.seatunnel.api.table.type.SeaTunnelDataType; | ||
import org.apache.seatunnel.api.table.type.SqlDateType; | ||
import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper; | ||
|
||
import lombok.extern.slf4j.Slf4j; | ||
|
||
import java.sql.ResultSetMetaData; | ||
import java.sql.SQLException; | ||
|
||
@Slf4j | ||
public class Gbase8aTypeMapper implements JdbcDialectTypeMapper { | ||
|
||
//ref http://www.gbase.cn/down/4419.html | ||
// ============================data types===================== | ||
private static final String GBASE8A_UNKNOWN = "UNKNOWN"; | ||
|
||
// -------------------------number---------------------------- | ||
private static final String GBASE8A_INT = "INT"; | ||
private static final String GBASE8A_TINYINT = "TINYINT"; | ||
private static final String GBASE8A_SMALLINT = "SMALLINT"; | ||
private static final String GBASE8A_BIGINT = "BIGINT"; | ||
private static final String GBASE8A_DECIMAL = "DECIMAL"; | ||
private static final String GBASE8A_FLOAT = "FLOAT"; | ||
private static final String GBASE8A_DOUBLE = "DOUBLE"; | ||
|
||
// -------------------------string---------------------------- | ||
private static final String GBASE8A_CHAR = "CHAR"; | ||
private static final String GBASE8A_VARCHAR = "VARCHAR"; | ||
|
||
|
||
// ------------------------------time------------------------- | ||
private static final String GBASE8A_DATE = "DATE"; | ||
private static final String GBASE8A_TIME = "TIME"; | ||
private static final String GBASE8A_TIMESTAMP = "TIMESTAMP"; | ||
private static final String GBASE8A_DATETIME = "DATETIME"; | ||
|
||
// ------------------------------blob------------------------- | ||
private static final String GBASE8A_BLOB = "BLOB"; | ||
private static final String GBASE8A_TEXT = "TEXT"; | ||
|
||
@SuppressWarnings("checkstyle:MagicNumber") | ||
@Override | ||
public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) throws SQLException { | ||
String gbase8aType = metadata.getColumnTypeName(colIndex).toUpperCase(); | ||
int precision = metadata.getPrecision(colIndex); | ||
int scale = metadata.getScale(colIndex); | ||
switch (gbase8aType) { | ||
case GBASE8A_TINYINT: | ||
return BasicType.BYTE_TYPE; | ||
case GBASE8A_SMALLINT: | ||
return BasicType.SHORT_TYPE; | ||
case GBASE8A_INT: | ||
return BasicType.INT_TYPE; | ||
case GBASE8A_BIGINT: | ||
return BasicType.LONG_TYPE; | ||
case GBASE8A_DECIMAL: | ||
if (precision < 38) { | ||
return new DecimalType(precision, scale); | ||
} | ||
return new DecimalType(38, 18); | ||
case GBASE8A_DOUBLE: | ||
return BasicType.DOUBLE_TYPE; | ||
case GBASE8A_FLOAT: | ||
return BasicType.FLOAT_TYPE; | ||
case GBASE8A_CHAR: | ||
case GBASE8A_VARCHAR: | ||
return BasicType.STRING_TYPE; | ||
case GBASE8A_DATE: | ||
return SqlDateType.SQL_DATE_TYPE; | ||
case GBASE8A_TIME: | ||
return SqlDateType.SQL_TIME_TYPE; | ||
case GBASE8A_TIMESTAMP: | ||
case GBASE8A_DATETIME: | ||
return SqlDateType.SQL_DATE_TIME_TYPE; | ||
case GBASE8A_BLOB: | ||
case GBASE8A_TEXT: | ||
return PrimitiveByteArrayType.INSTANCE; | ||
//Doesn't support yet | ||
case GBASE8A_UNKNOWN: | ||
default: | ||
final String jdbcColumnName = metadata.getColumnName(colIndex); | ||
throw new UnsupportedOperationException( | ||
String.format( | ||
"Doesn't support GBASE8A type '%s' on column '%s' yet.", | ||
gbase8aType, jdbcColumnName)); | ||
} | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -75,12 +75,11 @@ public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex) th | |
switch (oracleType) { | ||
case ORACLE_INTEGER: | ||
return BasicType.INT_TYPE; | ||
case ORACLE_FLOAT: | ||
case ORACLE_NUMBER: | ||
if (precision < 38) { | ||
return new DecimalType(precision, scale); | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can you add a comment about why delete this? |
||
//The float type will be converted to DecimalType(10, -127), | ||
// which will lose precision in the spark engine | ||
return new DecimalType(38, 18); | ||
case ORACLE_FLOAT: | ||
case ORACLE_BINARY_DOUBLE: | ||
return BasicType.DOUBLE_TYPE; | ||
case ORACLE_BINARY_FLOAT: | ||
|
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@ashulin PTAL about add new
SqlDateType
.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IMO, this is not necessary.
The same effect can be achieved using
LocalTimeType
.Multiple time types can confuse developers/users