github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-02 20:21:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #50 from taosdata/dev/zyyang

Browse Source 
fix: schema cache load column meta
This commit is contained in:
zyyang 2024-04-23 17:26:37 +08:00 committed by GitHub
commit 59d1e3ab1b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 15 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
tdengine30writer/src/main/java/com/alibaba/datax/plugin/writer/tdengine30writer/ColumnMeta.java
View File

@ -11,14 +11,12 @@ public class ColumnMeta {
@Override
public String toString() {
return "ColumnMeta{" +
"field='" + field + '\'' +
", type='" + type + '\'' +
", length=" + length +
", note='" + note + '\'' +
", isTag=" + isTag +
", isPrimaryKey=" + isPrimaryKey +
", value=" + value +
'}';
return "ColumnMeta{" + "field='" + field + '\'' + ", type='" + type + '\'' + ", length=" + length + ", note='" +
note + '\'' + ", isTag=" + isTag + ", isPrimaryKey=" + isPrimaryKey + ", value=" + value + '}';
}
@Override
public boolean equals(Object obj) {
return obj instanceof ColumnMeta && this.field.equals(((ColumnMeta) obj).field);
}
}

11
tdengine30writer/src/main/java/com/alibaba/datax/plugin/writer/tdengine30writer/SchemaCache.java
View File

@ -78,8 +78,10 @@ public final class SchemaCache {
synchronized (SchemaCache.class) {
if (columnMetas.get(tbname).isEmpty()) {
List<String> column_name = config.getList(Key.COLUMN, String.class);
List<ColumnMeta> colMetaList = getColumnMetaListFromDb(tbname,
(colMeta) -> column_name.contains(colMeta.field));
columnMetas.get(tbname).addAll(colMetaList);
}
}
@ -88,7 +90,7 @@ public final class SchemaCache {
}
private List<ColumnMeta> getColumnMetaListFromDb(String tableName, Predicate<ColumnMeta> filter) {
List<ColumnMeta> columnMetaList = columnMetas.get(tableName);
List<ColumnMeta> columnMetaList = new ArrayList<>();
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery("describe " + tableName);
@ -98,6 +100,7 @@ public final class SchemaCache {
if (filter.test(columnMeta))
columnMetaList.add(columnMeta);
}
rs.close();
} catch (SQLException e) {
throw DataXException.asDataXException(TDengineWriterErrorCode.RUNTIME_EXCEPTION, e.getMessage());
}
@ -117,8 +120,10 @@ public final class SchemaCache {
Object tagValue = null;
try (Statement stmt = conn.createStatement()) {
ResultSet rs = stmt.executeQuery(sql);
rs.next();
tagValue = rs.getObject(tagName);
while (rs.next()) {
tagValue = rs.getObject(tagName);
}
} catch (SQLException e) {
log.error("failed to get tag value, use NULL, cause: {" + e.getMessage() + "}");
}