database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[improvement](jdbc catalog) Delete unnecessary schema and optimize insert logic (#30880)

Browse Source 
In the previous design, we were compatible with MySQL's auto-increment column and default value to bypass the null value check when writing back Jdbc External Table. However, because MySQL's default value is not completely unified with Doris, this resulted in The unsuitable default value is wrong. In response to this situation, I made the following optimizations
1. For JDBC External Table, we always allow certain columns to be missing during insertion. Even if these columns are not allowed to be empty at the source end, the error should be generated by the source end, not Doris herself.
2. When the target column is non-nullable and the insertion is done via `INSERT INTO tbl VALUES()` or `INSERT INTO tbl SELECT constants`, Doris should verify any inconsistency between them and throw an exception. This check is not applied for `INSERT INTO tbl SELECT ... FROM tbl` operations.
This commit is contained in:
zy-kkk
2024-02-18 15:13:42 +08:00
committed by yiguolei
parent 3da168afc9
commit 6e4f76de54
6 changed files with 72 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
fe/fe-core/src/main/java/org/apache/doris/analysis/NativeInsertStmt.java
View File

@ -664,7 +664,10 @@ public class NativeInsertStmt extends InsertStmt {
}
// Check if all columns mentioned is enough
checkColumnCoverage(mentionedColumns, targetTable.getBaseSchema());
// For JdbcTable, it is allowed to insert without specifying all columns and without checking
if (!(targetTable instanceof JdbcTable)) {
checkColumnCoverage(mentionedColumns, targetTable.getBaseSchema());
}
realTargetColumnNames = targetColumns.stream().map(Column::getName).collect(Collectors.toList());
@ -675,6 +678,21 @@ public class NativeInsertStmt extends InsertStmt {
// INSERT INTO VALUES(...)
List<ArrayList<Expr>> rows = selectStmt.getValueList().getRows();
for (int rowIdx = 0; rowIdx < rows.size(); ++rowIdx) {
// Only check for JdbcTable
if (targetTable instanceof JdbcTable) {
// Check for NULL values in not-nullable columns
for (int colIdx = 0; colIdx < targetColumns.size(); ++colIdx) {
Column column = targetColumns.get(colIdx);
// Ensure rows.get(rowIdx) has enough columns to match targetColumns
if (colIdx < rows.get(rowIdx).size()) {
Expr expr = rows.get(rowIdx).get(colIdx);
if (!column.isAllowNull() && expr instanceof NullLiteral) {
throw new AnalysisException("Column `" + column.getName()
+ "` is not nullable, but the inserted value is nullable.");
}
}
}
}
analyzeRow(analyzer, targetColumns, rows, rowIdx, origColIdxsForExtendCols, realTargetColumnNames,
skipCheck);
}
@ -698,6 +716,19 @@ public class NativeInsertStmt extends InsertStmt {
skipCheck);
// rows may be changed in analyzeRow(), so rebuild the result exprs
selectStmt.getResultExprs().clear();
// For JdbcTable, need to check whether there is a NULL value inserted into the NOT NULL column
if (targetTable instanceof JdbcTable) {
for (int colIdx = 0; colIdx < targetColumns.size(); ++colIdx) {
Column column = targetColumns.get(colIdx);
Expr expr = rows.get(0).get(colIdx);
if (!column.isAllowNull() && expr instanceof NullLiteral) {
throw new AnalysisException("Column `" + column.getName()
+ "` is not nullable, but the inserted value is nullable.");
}
}
}
for (Expr expr : rows.get(0)) {
selectStmt.getResultExprs().add(expr);
}

6
fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcClient.java
View File

@ -325,7 +325,6 @@ public abstract class JdbcClient {
We used this method to retrieve the key column of the JDBC table, but since we only tested mysql,
we kept the default key behavior in the parent class and only overwrite it in the mysql subclass
*/
field.setKey(true);
field.setColumnSize(rs.getInt("COLUMN_SIZE"));
field.setDecimalDigits(rs.getInt("DECIMAL_DIGITS"));
field.setNumPrecRadix(rs.getInt("NUM_PREC_RADIX"));
@ -354,7 +353,7 @@ public abstract class JdbcClient {
List<Column> dorisTableSchema = Lists.newArrayListWithCapacity(jdbcTableSchema.size());
for (JdbcFieldSchema field : jdbcTableSchema) {
dorisTableSchema.add(new Column(field.getColumnName(),
jdbcTypeToDoris(field), field.isKey, null,
jdbcTypeToDoris(field), true, null,
field.isAllowNull(), field.getRemarks(),
true, -1));
}
@ -503,7 +502,6 @@ public abstract class JdbcClient {
protected int dataType;
// The SQL type of the corresponding java.sql.types (Type Name)
protected String dataTypeName;
protected boolean isKey;
// For CHAR/DATA, columnSize means the maximum number of chars.
// For NUMERIC/DECIMAL, columnSize means precision.
protected int columnSize;
@ -517,8 +515,6 @@ public abstract class JdbcClient {
// because for utf8 encoding, a Chinese character takes up 3 bytes
protected int charOctetLength;
protected boolean isAllowNull;
protected long autoIncInitValue;
protected String defaultValue;
}
protected abstract Type jdbcTypeToDoris(JdbcFieldSchema fieldSchema);

50
fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcMySQLClient.java
View File

@ -17,9 +17,7 @@
package org.apache.doris.datasource.jdbc.client;
import org.apache.doris.analysis.DefaultValueExprDef;
import org.apache.doris.catalog.ArrayType;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.Type;
@ -117,14 +115,13 @@ public class JdbcMySQLClient extends JdbcClient {
public List<JdbcFieldSchema> getJdbcColumnsInfo(String dbName, String tableName) {
Connection conn = getConnection();
ResultSet rs = null;
List<JdbcFieldSchema> tableSchema = com.google.common.collect.Lists.newArrayList();
List<JdbcFieldSchema> tableSchema = Lists.newArrayList();
String finalDbName = getRealDatabaseName(dbName);
String finalTableName = getRealTableName(dbName, tableName);
try {
DatabaseMetaData databaseMetaData = conn.getMetaData();
String catalogName = getCatalogName(conn);
rs = getColumns(databaseMetaData, catalogName, finalDbName, finalTableName);
List<String> primaryKeys = getPrimaryKeys(databaseMetaData, catalogName, dbName, tableName);
Map<String, String> mapFieldtoType = null;
while (rs.next()) {
lowerColumnToRealColumn.putIfAbsent(finalDbName, new ConcurrentHashMap<>());
@ -150,7 +147,6 @@ public class JdbcMySQLClient extends JdbcClient {
mapFieldtoType = getColumnsDataTypeUseQuery(dbName, tableName);
field.setDataTypeName(mapFieldtoType.get(rs.getString("COLUMN_NAME")));
}
field.setKey(primaryKeys.contains(field.getColumnName()));
field.setColumnSize(rs.getInt("COLUMN_SIZE"));
field.setDecimalDigits(rs.getInt("DECIMAL_DIGITS"));
field.setNumPrecRadix(rs.getInt("NUM_PREC_RADIX"));
@ -163,9 +159,6 @@ public class JdbcMySQLClient extends JdbcClient {
field.setAllowNull(rs.getInt("NULLABLE") != 0);
field.setRemarks(rs.getString("REMARKS"));
field.setCharOctetLength(rs.getInt("CHAR_OCTET_LENGTH"));
String isAutoincrement = rs.getString("IS_AUTOINCREMENT");
field.setAutoIncInitValue("YES".equalsIgnoreCase(isAutoincrement) ? 1 : -1);
field.setDefaultValue(rs.getString("COLUMN_DEF"));
tableSchema.add(field);
}
} catch (SQLException e) {
@ -177,47 +170,6 @@ public class JdbcMySQLClient extends JdbcClient {
return tableSchema;
}
@Override
public List<Column> getColumnsFromJdbc(String dbName, String tableName) {
List<JdbcFieldSchema> jdbcTableSchema = getJdbcColumnsInfo(dbName, tableName);
List<Column> dorisTableSchema = Lists.newArrayListWithCapacity(jdbcTableSchema.size());
for (JdbcFieldSchema field : jdbcTableSchema) {
DefaultValueExprDef defaultValueExprDef = null;
if (field.getDefaultValue() != null) {
String colDefaultValue = field.getDefaultValue().toLowerCase();
// current_timestamp()
if (colDefaultValue.startsWith("current_timestamp")) {
long precision = 0;
if (colDefaultValue.contains("(")) {
String substring = colDefaultValue.substring(18, colDefaultValue.length() - 1).trim();
precision = substring.isEmpty() ? 0 : Long.parseLong(substring);
}
defaultValueExprDef = new DefaultValueExprDef("now", precision);
}
}
dorisTableSchema.add(new Column(field.getColumnName(),
jdbcTypeToDoris(field), field.isKey(), null,
field.isAllowNull(), field.getAutoIncInitValue(), field.getDefaultValue(), field.getRemarks(),
true, defaultValueExprDef, -1, null));
}
return dorisTableSchema;
}
protected List<String> getPrimaryKeys(DatabaseMetaData databaseMetaData, String catalogName,
String dbName, String tableName) throws SQLException {
ResultSet rs = null;
List<String> primaryKeys = Lists.newArrayList();
rs = databaseMetaData.getPrimaryKeys(dbName, null, tableName);
while (rs.next()) {
String columnName = rs.getString("COLUMN_NAME");
primaryKeys.add(columnName);
}
rs.close();
return primaryKeys;
}
@Override
protected Type jdbcTypeToDoris(JdbcFieldSchema fieldSchema) {
// For Doris type

1
fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcOracleClient.java
View File

@ -134,7 +134,6 @@ public class JdbcOracleClient extends JdbcClient {
We used this method to retrieve the key column of the JDBC table, but since we only tested mysql,
we kept the default key behavior in the parent class and only overwrite it in the mysql subclass
*/
field.setKey(true);
field.setColumnSize(rs.getInt("COLUMN_SIZE"));
field.setDecimalDigits(rs.getInt("DECIMAL_DIGITS"));
field.setNumPrecRadix(rs.getInt("NUM_PREC_RADIX"));