You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2022/04/30 08:15:05 UTC
[shardingsphere] branch master updated: Make table name case-insensitive for PipelineTableMetaDataLoader (#17230)
This is an automated email from the ASF dual-hosted git repository.
zhaojinchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 73a87e81c6c Make table name case-insensitive for PipelineTableMetaDataLoader (#17230)
73a87e81c6c is described below
commit 73a87e81c6cbb7a27af16f24d1254526afc84cea
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Sat Apr 30 16:14:47 2022 +0800
Make table name case-insensitive for PipelineTableMetaDataLoader (#17230)
* Fix checkstyle
* Add more detailed log for metadata loading failed
* Add log to show create table SQLs for openGauss
* Make table name case-insensitive for PipelineTableMetaDataLoader
---
.../data/pipeline/api/metadata/TableName.java | 59 ++++++++++++++++++++++
.../check/consistency/DataConsistencyChecker.java | 2 +-
.../pipeline/core/importer/AbstractImporter.java | 6 ++-
.../loader/PipelineTableMetaDataLoader.java | 15 +++---
.../datasource/OpenGaussDataSourcePreparer.java | 1 +
5 files changed, 74 insertions(+), 9 deletions(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/metadata/TableName.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/metadata/TableName.java
new file mode 100644
index 00000000000..fbda35ec78d
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/metadata/TableName.java
@@ -0,0 +1,59 @@
+/*
+ * 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.shardingsphere.data.pipeline.api.metadata;
+
+import lombok.Getter;
+import lombok.NonNull;
+
+import java.util.Objects;
+
+/**
+ * Table name. It's case-insensitive.
+ */
+@Getter
+public final class TableName {
+
+ @NonNull
+ private final String lowercase;
+
+ public TableName(final String tableName) {
+ this.lowercase = tableName.toLowerCase();
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final TableName tableName = (TableName) o;
+ return lowercase.equals(tableName.lowercase);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(lowercase);
+ }
+
+ @Override
+ public String toString() {
+ return lowercase;
+ }
+}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
index 38370424eda..bc34be7fb6f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/check/consistency/DataConsistencyChecker.java
@@ -161,7 +161,7 @@ public final class DataConsistencyChecker {
logicTableNames.forEach(each -> {
// TODO put to preparer
if (!tableMetaDataMap.containsKey(each)) {
- throw new PipelineDataConsistencyCheckFailedException(String.format("Could not get metadata for table '%s'", each));
+ throw new PipelineDataConsistencyCheckFailedException(String.format("Could not get metadata for table '%s', tableMetaDataMap.keySet=%s", each, tableMetaDataMap.keySet()));
}
});
String sourceDatabaseType = sourceDataSourceConfig.getDatabaseType().getName();
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
index 3abf9d8f1ac..187e2f93d0f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/importer/AbstractImporter.java
@@ -178,7 +178,11 @@ public abstract class AbstractImporter extends AbstractLifecycleExecutor impleme
}
private void executeUpdate(final Connection connection, final DataRecord record) throws SQLException {
- List<Column> conditionColumns = RecordUtil.extractConditionColumns(record, importerConfig.getShardingColumnsMap().get(record.getTableName()));
+ Set<String> shardingColumns = importerConfig.getShardingColumnsMap().get(record.getTableName());
+ if (null == shardingColumns) {
+ log.error("executeUpdate, could not get shardingColumns, tableName={}, shardingColumnsMap.keySet={}", record.getTableName(), importerConfig.getShardingColumnsMap().keySet());
+ }
+ List<Column> conditionColumns = RecordUtil.extractConditionColumns(record, shardingColumns);
List<Column> updatedColumns = pipelineSqlBuilder.extractUpdatedColumns(record.getColumns(), record);
String updateSql = pipelineSqlBuilder.buildUpdateSQL(record, conditionColumns);
try (PreparedStatement ps = connection.prepareStatement(updateSql)) {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/loader/PipelineTableMetaDataLoader.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/loader/PipelineTableMetaDataLoader.java
index e5c9300c3d8..50898c5cdeb 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/loader/PipelineTableMetaDataLoader.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/metadata/loader/PipelineTableMetaDataLoader.java
@@ -20,6 +20,7 @@ package org.apache.shardingsphere.data.pipeline.core.metadata.loader;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
+import org.apache.shardingsphere.data.pipeline.api.metadata.TableName;
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineColumnMetaData;
import org.apache.shardingsphere.data.pipeline.core.metadata.model.PipelineTableMetaData;
@@ -42,7 +43,7 @@ public final class PipelineTableMetaDataLoader {
private final PipelineDataSourceWrapper dataSource;
- private final Map<String, PipelineTableMetaData> tableMetaDataMap = new ConcurrentHashMap<>();
+ private final Map<TableName, PipelineTableMetaData> tableMetaDataMap = new ConcurrentHashMap<>();
/**
* Load table metadata.
@@ -53,13 +54,13 @@ public final class PipelineTableMetaDataLoader {
public void loadTableMetaData(final String tableNamePattern) throws SQLException {
try (Connection connection = dataSource.getConnection()) {
long startMillis = System.currentTimeMillis();
- Map<String, PipelineTableMetaData> tableMetaDataMap = loadTableMetaData0(connection, tableNamePattern);
+ Map<TableName, PipelineTableMetaData> tableMetaDataMap = loadTableMetaData0(connection, tableNamePattern);
log.info("loadTableMetaData, tableNamePattern={}, result={}, cost time={} ms", tableNamePattern, tableMetaDataMap, System.currentTimeMillis() - startMillis);
this.tableMetaDataMap.putAll(tableMetaDataMap);
}
}
- private Map<String, PipelineTableMetaData> loadTableMetaData0(final Connection connection, final String tableNamePattern) throws SQLException {
+ private Map<TableName, PipelineTableMetaData> loadTableMetaData0(final Connection connection, final String tableNamePattern) throws SQLException {
Map<String, Map<String, PipelineColumnMetaData>> tablePipelineColumnMetaDataMap = new LinkedHashMap<>();
// TODO if tableNamePattern is '%', it might return inconsistent result, actual table `t_order_2` may be return
try (ResultSet resultSet = connection.getMetaData().getColumns(connection.getCatalog(), null, tableNamePattern, "%")) {
@@ -85,9 +86,9 @@ public final class PipelineTableMetaDataLoader {
columnMetaDataMap.put(columnName, columnMetaData);
}
}
- Map<String, PipelineTableMetaData> result = new LinkedHashMap<>();
+ Map<TableName, PipelineTableMetaData> result = new LinkedHashMap<>();
for (Entry<String, Map<String, PipelineColumnMetaData>> entry : tablePipelineColumnMetaDataMap.entrySet()) {
- result.put(entry.getKey(), new PipelineTableMetaData(entry.getKey(), entry.getValue()));
+ result.put(new TableName(entry.getKey()), new PipelineTableMetaData(entry.getKey(), entry.getValue()));
}
return result;
}
@@ -110,7 +111,7 @@ public final class PipelineTableMetaDataLoader {
* @return table metadata
*/
public PipelineTableMetaData getTableMetaData(final String tableName) {
- PipelineTableMetaData result = tableMetaDataMap.get(tableName);
+ PipelineTableMetaData result = tableMetaDataMap.get(new TableName(tableName));
if (null != result) {
return result;
}
@@ -119,7 +120,7 @@ public final class PipelineTableMetaDataLoader {
} catch (final SQLException ex) {
throw new RuntimeException(String.format("Load metadata for table '%s' failed", tableName), ex);
}
- result = tableMetaDataMap.get(tableName);
+ result = tableMetaDataMap.get(new TableName(tableName));
if (null == result) {
log.warn("getTableMetaData, can not load metadata for table '{}'", tableName);
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
index 0dcdf686db0..86b9234f20f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-opengauss/src/main/java/org/apache/shardingsphere/data/pipeline/opengauss/prepare/datasource/OpenGaussDataSourcePreparer.java
@@ -86,6 +86,7 @@ public final class OpenGaussDataSourcePreparer extends AbstractDataSourcePrepare
String actualTableName = dataNode.getTableName();
int oid = queryTableOid(sourceConnection, actualTableName);
String tableDefinition = queryTableDefinition(sourceConnection, oid);
+ log.info("getActualTableDefinitions, dataNode={}, tableDefinition={}", dataNode, tableDefinition);
String logicTableName = each.getLogicTableName();
result.add(new ActualTableDefinition(logicTableName, actualTableName, tableDefinition));
}