You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by xx...@apache.org on 2022/12/05 10:21:02 UTC
[kylin] 10/22: skip secondary index while modify column
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 13bd871bdf9d18bb2431cba466b80b6244ba607f
Author: Zhixiong Chen <ch...@apache.org>
AuthorDate: Fri Oct 14 12:56:10 2022 +0800
skip secondary index while modify column
---
.../kyligence/kap/secondstorage/SecondStorageLockTest.java | 14 +++++++++++++-
.../kap/clickhouse/database/ClickHouseQueryOperator.java | 8 ++++++--
.../management/SecondStorageScheduleService.java | 7 ++++++-
.../kap/secondstorage/management/SecondStorageService.java | 6 ++++++
.../kap/secondstorage/database/QueryOperator.java | 3 ++-
5 files changed, 33 insertions(+), 5 deletions(-)
diff --git a/src/second-storage/clickhouse-it/src/test/java/io/kyligence/kap/secondstorage/SecondStorageLockTest.java b/src/second-storage/clickhouse-it/src/test/java/io/kyligence/kap/secondstorage/SecondStorageLockTest.java
index 03c191ab16..ffb4340233 100644
--- a/src/second-storage/clickhouse-it/src/test/java/io/kyligence/kap/secondstorage/SecondStorageLockTest.java
+++ b/src/second-storage/clickhouse-it/src/test/java/io/kyligence/kap/secondstorage/SecondStorageLockTest.java
@@ -2867,7 +2867,7 @@ public class SecondStorageLockTest implements JobWaiter {
int replica = 1;
configClickhouseWith(clickhouse, replica, catalog, () -> {
QueryOperator queryOperator = SecondStorageFactoryUtils.createQueryMetricOperator(getProject());
- queryOperator.modifyColumnByCardinality("default", "table");
+ queryOperator.modifyColumnByCardinality("default", "table", Sets.newHashSet());
buildIncrementalLoadQuery("2012-01-02", "2012-01-03");
waitAllJobFinish();
@@ -2938,6 +2938,18 @@ public class SecondStorageLockTest implements JobWaiter {
}
}
assertEquals(LOW_CARDINALITY_STRING, rows);
+
+ queryOperator.modifyColumnByCardinality(database, destTableName, Sets.newHashSet(4));
+ try (Connection connection = DriverManager.getConnection(clickhouse1.getJdbcUrl());
+ val stmt = connection.createStatement()) {
+ val rs = stmt.executeQuery(String.format(Locale.ROOT, "desc %s.%s", database, destTableName));
+ while (rs.next()) {
+ if ("c4".equals(rs.getString(1))) {
+ rows = rs.getString(2);
+ }
+ }
+ }
+ assertEquals(LOW_CARDINALITY_STRING, rows);
return true;
});
}
diff --git a/src/second-storage/clickhouse/src/main/java/io/kyligence/kap/clickhouse/database/ClickHouseQueryOperator.java b/src/second-storage/clickhouse/src/main/java/io/kyligence/kap/clickhouse/database/ClickHouseQueryOperator.java
index 7e0b37d3bd..ab214644ba 100644
--- a/src/second-storage/clickhouse/src/main/java/io/kyligence/kap/clickhouse/database/ClickHouseQueryOperator.java
+++ b/src/second-storage/clickhouse/src/main/java/io/kyligence/kap/clickhouse/database/ClickHouseQueryOperator.java
@@ -28,6 +28,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import io.kyligence.kap.secondstorage.ColumnMapping;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -166,7 +167,7 @@ public class ClickHouseQueryOperator implements QueryOperator {
QueryMetrics.SOURCE_RESULT_COUNT, -1L);
}
- public void modifyColumnByCardinality(String database, String destTableName) {
+ public void modifyColumnByCardinality(String database, String destTableName, Set<Integer> secondaryIndex) {
KylinConfig config = KylinConfig.getInstanceFromEnv();
List<NodeGroup> nodeGroups = SecondStorageUtil.listNodeGroup(config, project);
Set<String> nodes = nodeGroups.stream()
@@ -178,7 +179,10 @@ public class ClickHouseQueryOperator implements QueryOperator {
return;
ProjectInstance projectInstance = NProjectManager.getInstance(config).getProject(project);
- List<ClickHouseSystemQuery.DescTable> modifyColumns = getFilterDescTable(maxRowsNode, database, destTableName, projectInstance.getConfig());
+ val tableColumns = getFilterDescTable(maxRowsNode, database, destTableName, projectInstance.getConfig());
+ val modifyColumns = tableColumns.stream()
+ .filter(col -> !secondaryIndex.contains(Integer.valueOf(ColumnMapping.secondStorageColumnToKapColumn(col.getColumn()))))
+ .collect(Collectors.toList());
if (CollectionUtils.isEmpty(modifyColumns))
return;
diff --git a/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageScheduleService.java b/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageScheduleService.java
index 97ac724418..eec15691e2 100644
--- a/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageScheduleService.java
+++ b/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageScheduleService.java
@@ -19,6 +19,8 @@
package io.kyligence.kap.secondstorage.management;
import com.google.common.collect.Maps;
+import io.kyligence.kap.secondstorage.metadata.TableEntity;
+import io.kyligence.kap.secondstorage.metadata.TablePlan;
import org.apache.kylin.metadata.cube.model.NDataflowManager;
import org.apache.kylin.metadata.epoch.EpochManager;
import org.apache.kylin.metadata.model.NDataModel;
@@ -98,7 +100,10 @@ public class SecondStorageScheduleService {
try {
val database = NameUtil.getDatabase(df);
val destTableName = NameUtil.getTable(df, SecondStorageUtil.getBaseIndex(df).getId());
- queryOperator.modifyColumnByCardinality(database, destTableName);
+ val tablePlanManager = SecondStorageUtil.tablePlanManager(config, project);
+ TablePlan tablePlan = tablePlanManager.get().get(model.getId()).get();
+ TableEntity tableEntity = tablePlan.getEntity(SecondStorageUtil.getBaseIndex(df).getId()).orElse(null);
+ queryOperator.modifyColumnByCardinality(database, destTableName, tableEntity.getSecondaryIndexColumns());
} catch (Exception exception) {
log.error("Failed to modify second storage low cardinality on model {}.", model.getId(), exception);
} finally {
diff --git a/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageService.java b/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageService.java
index 460e40faed..b62b4e2376 100644
--- a/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageService.java
+++ b/src/second-storage/core-ui/src/main/java/io/kyligence/kap/secondstorage/management/SecondStorageService.java
@@ -1196,6 +1196,12 @@ public class SecondStorageService extends BasicService implements SecondStorageU
if (StringUtils.isEmpty(colPrefix.get()))
throw new KylinException(INVALID_PARAMETER, String.format("There is no column %s in model %s", column, df.getModel().getAlias()));
+ val tablePlanManager = SecondStorageUtil.tablePlanManager(config, project);
+ TablePlan tablePlan = tablePlanManager.get().get(model).get();
+ TableEntity tableEntity = tablePlan.getEntity(SecondStorageUtil.getBaseIndex(df).getId()).orElse(null);
+ if (tableEntity.getSecondaryIndexColumns().contains(Integer.valueOf(ColumnMapping.secondStorageColumnToKapColumn(colPrefix.get()))))
+ throw new KylinException(INVALID_PARAMETER, String.format("The column %s is Secondary Index Column.", column));
+
val destTableName = NameUtil.getTable(df, layout.getId());
queryOperator.modifyColumnByCardinality(database, destTableName, colPrefix.get(), datatype);
} catch (Exception exception) {
diff --git a/src/second-storage/core/src/main/java/io/kyligence/kap/secondstorage/database/QueryOperator.java b/src/second-storage/core/src/main/java/io/kyligence/kap/secondstorage/database/QueryOperator.java
index a723e6f337..c0f04b65b6 100644
--- a/src/second-storage/core/src/main/java/io/kyligence/kap/secondstorage/database/QueryOperator.java
+++ b/src/second-storage/core/src/main/java/io/kyligence/kap/secondstorage/database/QueryOperator.java
@@ -19,11 +19,12 @@
package io.kyligence.kap.secondstorage.database;
import java.util.Map;
+import java.util.Set;
public interface QueryOperator {
Map<String, Object> getQueryMetric(String queryId);
- void modifyColumnByCardinality(String database, String destTableName);
+ void modifyColumnByCardinality(String database, String destTableName, Set<Integer> secondaryIndex);
void modifyColumnByCardinality(String database, String destTableName, String column, String datatype);
}