dataList) throws CheckConsumerPollEmptyException {
-
-
- }
-
/**
*
* Align the bucket list data according to the statistical results of source
@@ -411,12 +396,10 @@ public class SliceCheckWorker implements Runnable {
if (MapUtils.isNotEmpty(bucketDiff)) {
bucketDiff.forEach((number, pair) -> {
if (pair.getSource() == -1) {
- sourceTuple.getBuckets()
- .add(BuilderBucketHandler.builderEmpty(number));
+ sourceTuple.getBuckets().add(BuilderBucketHandler.builderEmpty(number));
}
if (pair.getSink() == -1) {
- sinkTuple.getBuckets()
- .add(BuilderBucketHandler.builderEmpty(number));
+ sinkTuple.getBuckets().add(BuilderBucketHandler.builderEmpty(number));
}
});
}
diff --git a/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/PrimaryColumnBean.java b/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/PrimaryColumnBean.java
index 651e26e27bfb2e4a8acdd6299951b18fb4caba05..5f85c51609f5d9fbe231a06536e874629c601be2 100644
--- a/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/PrimaryColumnBean.java
+++ b/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/PrimaryColumnBean.java
@@ -15,7 +15,9 @@
package org.opengauss.datachecker.common.entry.extract;
+import lombok.AllArgsConstructor;
import lombok.Data;
+import lombok.NoArgsConstructor;
/**
* PrimaryColumnBean
@@ -25,6 +27,8 @@ import lombok.Data;
* @since ：11
*/
@Data
+@NoArgsConstructor
+@AllArgsConstructor
public class PrimaryColumnBean {
/**
* Table
diff --git a/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/UniqueColumnBean.java b/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/UniqueColumnBean.java
new file mode 100644
index 0000000000000000000000000000000000000000..158f14723fd3246532768c66c9ed295c7e3f7ee8
--- /dev/null
+++ b/datachecker-common/src/main/java/org/opengauss/datachecker/common/entry/extract/UniqueColumnBean.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
+ *
+ * openGauss is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ *
+ * http://license.coscl.org.cn/MulanPSL2
+ *
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ */
+
+package org.opengauss.datachecker.common.entry.extract;
+
+import lombok.Data;
+
+/**
+ * UniqueColumnBean
+ *
+ * @author ：wangchao
+ * @date ：Created in 2023/12/23
+ * @since ：11
+ */
+@Data
+public class UniqueColumnBean {
+ /**
+ * Table
+ */
+ private String tableName;
+
+ /**
+ * Primary key column name
+ */
+ private String columnName;
+
+ /**
+ * Index identifier
+ */
+ private String indexIdentifier;
+
+ /**
+ * Column index
+ */
+ private Integer colIdx;
+}
\ No newline at end of file
diff --git a/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/BaseDataService.java b/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/BaseDataService.java
index 45d3c7ceb391166618bc96037975e5446eaed18e..54e94514887f51aa8dd5b9189cbf0303b5d50f74 100644
--- a/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/BaseDataService.java
+++ b/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/BaseDataService.java
@@ -16,6 +16,7 @@
package org.opengauss.datachecker.extract.data;
import com.alibaba.druid.pool.DruidDataSource;
+
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.logging.log4j.Logger;
@@ -33,6 +34,7 @@ import org.opengauss.datachecker.extract.service.RuleAdapterService;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
+
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
@@ -103,15 +105,13 @@ public class BaseDataService {
*/
public List bdsQueryTableMetadataList() {
List metadataList = dataAccessService.dasQueryTableMetadataList();
- return metadataList.stream()
- .filter(meta -> {
- boolean isChecking = ruleAdapterService.filterTableByRule(meta.getTableName());
- if (isChecking) {
- tableNameList.add(meta.getTableName());
- }
- return isChecking;
- })
- .collect(Collectors.toList());
+ return metadataList.stream().filter(meta -> {
+ boolean isChecking = ruleAdapterService.filterTableByRule(meta.getTableName());
+ if (isChecking) {
+ tableNameList.add(meta.getTableName());
+ }
+ return isChecking;
+ }).collect(Collectors.toList());
}
/**
@@ -124,8 +124,7 @@ public class BaseDataService {
if (CollectionUtils.isEmpty(columnBeanList)) {
return new HashMap<>();
}
- return columnBeanList.stream()
- .collect(Collectors.groupingBy(PrimaryColumnBean::getTableName));
+ return columnBeanList.stream().collect(Collectors.groupingBy(PrimaryColumnBean::getTableName));
}
private List filterByTableRules(List tableNameList) {
@@ -189,28 +188,31 @@ public class BaseDataService {
/**
* update table metadata, and filter column rules
*
- * @param tableMetadata table metadata
+ * @param tableMetadata table metadata
* @param primaryColumnBeans primary column
*/
public void updateTableColumnMetaData(TableMetadata tableMetadata, List primaryColumnBeans) {
String tableName = tableMetadata.getTableName();
final List columns = dataAccessService.queryTableColumnsMetaData(tableName);
- if (Objects.isNull(columns)) {
+ if (CollectionUtils.isEmpty(columns)) {
LogUtils.error(log, "table columns metadata is null ,{}", tableName);
return;
}
- if (Objects.isNull(primaryColumnBeans)) {
- primaryColumnBeans = dataAccessService.queryTablePrimaryColumns(tableName);
+ List tempPrimaryColumnBeans = primaryColumnBeans;
+ if (CollectionUtils.isEmpty(primaryColumnBeans)) {
+ tempPrimaryColumnBeans = dataAccessService.queryTablePrimaryColumns(tableName);
}
- if (Objects.nonNull(primaryColumnBeans)) {
- List primaryColumnNameList = getPrimaryColumnNames(primaryColumnBeans);
+ if (CollectionUtils.isEmpty(tempPrimaryColumnBeans)) {
+ tempPrimaryColumnBeans = dataAccessService.queryTableUniqueColumns(tableName);
+ }
+ if (CollectionUtils.isNotEmpty(tempPrimaryColumnBeans)) {
+ List primaryColumnNameList = getPrimaryColumnNames(tempPrimaryColumnBeans);
for (ColumnsMetaData column : columns) {
if (primaryColumnNameList.contains(column.getLowerCaseColumnName())) {
column.setColumnKey(ColumnKey.PRI);
}
}
}
-
tableMetadata.setColumnsMetas(ruleAdapterService.executeColumnRule(columns));
tableMetadata.setPrimaryMetas(getTablePrimaryColumn(columns));
tableMetadata.setTableHash(calcTableHash(columns));
@@ -218,16 +220,17 @@ public class BaseDataService {
private List getPrimaryColumnNames(List primaryColumnBeans) {
return primaryColumnBeans.stream()
- .map(PrimaryColumnBean::getColumnName)
- .map(String::toLowerCase)
- .collect(Collectors.toList());
+ .map(PrimaryColumnBean::getColumnName)
+ .map(String::toLowerCase)
+ .distinct()
+ .collect(Collectors.toList());
}
private List getTablePrimaryColumn(List columnsMetaData) {
return columnsMetaData.stream()
- .filter(meta -> ColumnKey.PRI.equals(meta.getColumnKey()))
- .sorted(Comparator.comparing(ColumnsMetaData::getOrdinalPosition))
- .collect(Collectors.toList());
+ .filter(meta -> ColumnKey.PRI.equals(meta.getColumnKey()))
+ .sorted(Comparator.comparing(ColumnsMetaData::getOrdinalPosition))
+ .collect(Collectors.toList());
}
/**
@@ -255,9 +258,8 @@ public class BaseDataService {
private long calcTableHash(List columnsMetas) {
StringBuilder buffer = new StringBuilder();
columnsMetas.sort(Comparator.comparing(ColumnsMetaData::getOrdinalPosition));
- columnsMetas.forEach(column -> buffer.append(column.getColumnName()
- .toLowerCase(Locale.ENGLISH))
- .append(column.getOrdinalPosition()));
+ columnsMetas.forEach(column -> buffer.append(column.getColumnName().toLowerCase(Locale.ENGLISH))
+ .append(column.getOrdinalPosition()));
return HASH_UTIL.hashBytes(buffer.toString());
}
@@ -289,9 +291,8 @@ public class BaseDataService {
} else {
String[] sqlModeArray = sqlMode.split(",");
String newSqlMode = Arrays.stream(sqlModeArray)
- .filter(mode -> !mode.equalsIgnoreCase(
- ConfigConstants.SQL_MODE_NAME_PAD_CHAR_TO_FULL_LENGTH))
- .collect(Collectors.joining(","));
+ .filter(mode -> !mode.equalsIgnoreCase(ConfigConstants.SQL_MODE_NAME_PAD_CHAR_TO_FULL_LENGTH))
+ .collect(Collectors.joining(","));
boolean isPadCharFull = ConfigCache.getBooleanValue(ConfigConstants.SQL_MODE_PAD_CHAR_TO_FULL_LENGTH);
if (isPadCharFull) {
newSqlMode += ConfigConstants.SQL_MODE_NAME_PAD_CHAR_TO_FULL_LENGTH;
diff --git a/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/access/AbstractDataAccessService.java b/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/access/AbstractDataAccessService.java
index a83d49811f0befc52ae145cf1aef556d7b4c4c9f..93b19b87bd7e11fc9da9aab42a3a3be1e4a55f37 100644
--- a/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/access/AbstractDataAccessService.java
+++ b/datachecker-extract/src/main/java/org/opengauss/datachecker/extract/data/access/AbstractDataAccessService.java
@@ -16,6 +16,9 @@
package org.opengauss.datachecker.extract.data.access;
import com.alibaba.druid.pool.DruidDataSource;
+
+import cn.hutool.core.collection.CollUtil;
+
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
import org.opengauss.datachecker.common.config.ConfigCache;
@@ -25,6 +28,7 @@ import org.opengauss.datachecker.common.entry.common.Health;
import org.opengauss.datachecker.common.entry.enums.LowerCaseTableNames;
import org.opengauss.datachecker.common.entry.extract.PrimaryColumnBean;
import org.opengauss.datachecker.common.entry.extract.TableMetadata;
+import org.opengauss.datachecker.common.entry.extract.UniqueColumnBean;
import org.opengauss.datachecker.common.exception.ExtractDataAccessException;
import org.opengauss.datachecker.common.util.DurationUtils;
import org.opengauss.datachecker.common.util.LogUtils;
@@ -36,16 +40,19 @@ import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import javax.annotation.Resource;
import javax.sql.DataSource;
+
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.time.LocalDateTime;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.stream.Collectors;
/**
* AbstractDataAccessService
@@ -113,7 +120,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
public String adasQuerySchema(Connection connection, String executeQueryStatement) {
String schema = "";
try (PreparedStatement ps = connection.prepareStatement(executeQueryStatement);
- ResultSet resultSet = ps.executeQuery()) {
+ ResultSet resultSet = ps.executeQuery()) {
if (resultSet.next()) {
schema = resultSet.getString(RS_COL_SCHEMA);
}
@@ -129,7 +136,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
* 数据库schema是否合法
*
* @param schema schema
- * @param sql sql
+ * @param sql sql
* @return result
*/
public Health health(String schema, String sql) {
@@ -160,7 +167,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
Connection connection = getConnection();
List list = new LinkedList<>();
try (PreparedStatement ps = connection.prepareStatement(executeQueryStatement);
- ResultSet resultSet = ps.executeQuery()) {
+ ResultSet resultSet = ps.executeQuery()) {
while (resultSet.next()) {
list.add(resultSet.getString(RS_COL_TABLE_NAME));
}
@@ -185,7 +192,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
Connection connection = getConnection();
List list = new LinkedList<>();
try (PreparedStatement ps = connection.prepareStatement(executeQueryStatement);
- ResultSet resultSet = ps.executeQuery()) {
+ ResultSet resultSet = ps.executeQuery()) {
PrimaryColumnBean metadata;
while (resultSet.next()) {
metadata = new PrimaryColumnBean();
@@ -203,6 +210,50 @@ public abstract class AbstractDataAccessService implements DataAccessService {
return list;
}
+ /**
+ * adas查询表的唯一性约束列信息
+ *
+ * @param executeQueryStatement executeQueryStatement
+ * @return List
+ */
+ public List adasQueryTableUniqueColumns(String executeQueryStatement) {
+ Connection connection = getConnection();
+ List list = new LinkedList<>();
+ try (PreparedStatement ps = connection.prepareStatement(executeQueryStatement);
+ ResultSet resultSet = ps.executeQuery()) {
+ UniqueColumnBean metadata;
+ while (resultSet.next()) {
+ metadata = new UniqueColumnBean();
+ metadata.setTableName(resultSet.getString("tableName"));
+ metadata.setColumnName(resultSet.getString("columnName"));
+ metadata.setIndexIdentifier(resultSet.getString("indexIdentifier"));
+ metadata.setColIdx(resultSet.getInt("colIdx"));
+ list.add(metadata);
+ }
+ } catch (SQLException esql) {
+ LogUtils.error(log, "adasQueryTablePrimaryColumns error:", esql);
+ } finally {
+ closeConnection(connection);
+ }
+ return list;
+ }
+
+ /**
+ * 将UniqueColumnBean列表转换为PrimaryColumnBean列表
+ *
+ * @param uniqueColumns 输入的UniqueColumnBean列表，可能为空
+ * @return PrimaryColumnBean列表，永远不会为null，其中的元素是唯一的
+ */
+ public List translateUniqueToPrimaryColumns(List uniqueColumns) {
+ if (CollUtil.isEmpty(uniqueColumns)) {
+ return new ArrayList<>();
+ }
+ return uniqueColumns.stream()
+ .map(u -> new PrimaryColumnBean(u.getTableName(), u.getColumnName()))
+ .distinct()
+ .collect(Collectors.toList());
+ }
+
/**
* adasQueryTableMetadataList
*
@@ -214,7 +265,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
Connection connection = getConnection();
List list = new LinkedList<>();
try (PreparedStatement ps = connection.prepareStatement(executeQueryStatement);
- ResultSet resultSet = ps.executeQuery()) {
+ ResultSet resultSet = ps.executeQuery()) {
TableMetadata metadata;
while (resultSet.next()) {
metadata = new TableMetadata();
@@ -238,7 +289,7 @@ public abstract class AbstractDataAccessService implements DataAccessService {
* 查询表数据抽样检查点清单
*
* @param connection connection
- * @param sql 检查点查询SQL
+ * @param sql 检查点查询SQL
* @return 检查点列表
*/
protected List