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 2023/08/25 09:08:43 UTC
[kylin] 02/03: Remove Second Storage, Step1
This is an automated email from the ASF dual-hosted git repository.
xxyu pushed a commit to branch release-pipeline
in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 2fc92b8be4eb70eb849c8d751c25bdcaecb4073d
Author: XiaoxiangYu <xx...@apache.org>
AuthorDate: Fri Aug 25 16:20:23 2023 +0800
Remove Second Storage, Step1
There are two part of source code, Kylin code and Second Storage code,
Kylin code will refer Second Storage code.
Step1: remove the reference from Kylin code to Second Storage code
Step2: remove the Second Storage in all Kylin code's pom.xml
Step3: just delete all Second Storage modules
Step4: run and passed UT&IT
Step5: remove the commented code in Step1&2
---
.../kylin/rest/response/ProjectConfigResponse.java | 6 +-
.../apache/kylin/rest/service/ProjectService.java | 18 +-
.../apache/kylin/rest/service/ServiceTestBase.java | 14 +-
.../org/apache/kylin/rest/service/JobService.java | 23 +-
.../apache/kylin/rest/service/JobServiceTest.java | 16 +-
.../kylin/rest/response/NDataModelResponse.java | 6 +-
.../kylin/rest/response/NDataSegmentResponse.java | 8 +-
.../apache/kylin/rest/service/ModelService.java | 452 ++-
.../org/apache/kylin/rest/util/ModelUtils.java | 60 +-
.../org/apache/kylin/rest/util/ModelUtilsTest.java | 64 +-
.../apache/kylin/query/relnode/OLAPContext.java | 24 +-
.../org/apache/kylin/query/routing/Candidate.java | 32 +-
.../kylin/rest/service/ModelQueryService.java | 12 +-
.../kylin/rest/service/QueryHistoryScheduler.java | 72 +-
.../kylin/rest/service/ModelQueryServiceTest.java | 58 +-
.../kap/secondstorage/SecondStorageLockTest.java | 3024 --------------------
.../management/SecondStorageService.java | 2 +-
.../spark/job/NSparkCleanupAfterMergeStep.java | 12 +-
.../kylin/engine/spark/job/NSparkCubingJob.java | 158 +-
.../kylin/engine/spark/job/NSparkMergingJob.java | 74 +-
.../kylin/query/runtime/plan/ResultPlan.scala | 14 +-
.../apache/spark/sql/KylinDataFrameManager.scala | 6 +-
.../kylin/tool/AbstractInfoExtractorTool.java | 2 +-
.../org/apache/kylin/tool/ClickhouseDiagTool.java | 240 --
.../apache/kylin/tool/ClickhouseDiagToolTest.java | 212 --
25 files changed, 560 insertions(+), 4049 deletions(-)
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java b/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
index 613915a7ad..029fd505cb 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/response/ProjectConfigResponse.java
@@ -27,7 +27,7 @@ import org.apache.kylin.metadata.model.VolatileRange;
import com.fasterxml.jackson.annotation.JsonProperty;
-import io.kyligence.kap.secondstorage.response.SecondStorageNode;
+//import io.kyligence.kap.secondstorage.response.SecondStorageNode;
import lombok.Getter;
import lombok.Setter;
@@ -70,8 +70,8 @@ public class ProjectConfigResponse {
@JsonProperty("second_storage_enabled")
private Boolean secondStorageEnabled;
- @JsonProperty("second_storage_nodes")
- private List<SecondStorageNode> secondStorageNodes;
+// @JsonProperty("second_storage_nodes")
+// private List<SecondStorageNode> secondStorageNodes;
@JsonProperty("runner_class_name")
private String runnerClassName;
@JsonProperty("converter_class_names")
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java b/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
index c3d1b8d0b6..f8782fdd16 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/service/ProjectService.java
@@ -140,7 +140,7 @@ import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.SneakyThrows;
import lombok.val;
@@ -611,10 +611,10 @@ public class ProjectService extends BasicService {
response.setJdbcSourceEnable(config.getJdbcEnable());
response.setJdbcSourceDriver(config.getJdbcDriver());
- if (SecondStorageUtil.isGlobalEnable()) {
- response.setSecondStorageEnabled(SecondStorageUtil.isProjectEnable(project));
- response.setSecondStorageNodes(SecondStorageUtil.listProjectNodes(project));
- }
+// if (SecondStorageUtil.isGlobalEnable()) {
+// response.setSecondStorageEnabled(SecondStorageUtil.isProjectEnable(project));
+// response.setSecondStorageNodes(SecondStorageUtil.listProjectNodes(project));
+// }
return response;
}
@@ -892,10 +892,10 @@ public class ProjectService extends BasicService {
@PreAuthorize(Constant.ACCESS_HAS_ROLE_ADMIN)
@Transaction(project = 0)
public void dropProject(String project) {
- if (SecondStorageUtil.isProjectEnable(project)) {
- throw new KylinException(PROJECT_DROP_FAILED,
- String.format(Locale.ROOT, MsgPicker.getMsg().getProjectDropFailedSecondStorageEnabled(), project));
- }
+// if (SecondStorageUtil.isProjectEnable(project)) {
+// throw new KylinException(PROJECT_DROP_FAILED,
+// String.format(Locale.ROOT, MsgPicker.getMsg().getProjectDropFailedSecondStorageEnabled(), project));
+// }
val kylinConfig = KylinConfig.getInstanceFromEnv();
NExecutableManager nExecutableManager = NExecutableManager.getInstance(kylinConfig, project);
diff --git a/src/common-service/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java b/src/common-service/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
index f6fcc3af3b..e8666ce866 100644
--- a/src/common-service/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
+++ b/src/common-service/src/test/java/org/apache/kylin/rest/service/ServiceTestBase.java
@@ -31,11 +31,11 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.Mockito;
+//import org.mockito.Mockito;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Bean;
+//import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
@@ -49,7 +49,7 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
-import io.kyligence.kap.secondstorage.SecondStorageUpdater;
+//import io.kyligence.kap.secondstorage.SecondStorageUpdater;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = ServiceTestBase.SpringConfig.class)
@@ -118,9 +118,9 @@ public class ServiceTestBase extends NLocalFileMetadataTestCase {
@ImportResource(locations = { "classpath:applicationContext.xml", "classpath:kylinSecurity.xml" })
@EnableAsync
public static class SpringConfig {
- @Bean
- public SecondStorageUpdater getSecondStorageUpdater() {
- return Mockito.mock(SecondStorageUpdater.class, invocationOnMock -> "");
- }
+// @Bean
+// public SecondStorageUpdater getSecondStorageUpdater() {
+// return Mockito.mock(SecondStorageUpdater.class, invocationOnMock -> "");
+// }
}
}
diff --git a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
index ca2664ec31..e03a16ae72 100644
--- a/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
+++ b/src/data-loading-service/src/main/java/org/apache/kylin/rest/service/JobService.java
@@ -140,7 +140,7 @@ import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
@@ -277,15 +277,12 @@ public class JobService extends BasicService implements JobSupporter, ISmartAppl
}
public List<ExecutableResponse.SegmentResponse> getSegments(AbstractExecutable executable) {
- if (SecondStorageUtil.isModelEnable(executable.getProject(), executable.getTargetModelId())) {
- return modelService
- .getSegmentsResponseByJob(executable.getTargetModelId(), executable.getProject(), executable)
- .stream()
- .map(dataSegmentResponse -> new ExecutableResponse.SegmentResponse(dataSegmentResponse.getId(),
- dataSegmentResponse.getStatusToDisplay()))
- .collect(Collectors.toList());
- }
- return Lists.newArrayList();
+ return modelService
+ .getSegmentsResponseByJob(executable.getTargetModelId(), executable.getProject(), executable)
+ .stream()
+ .map(dataSegmentResponse -> new ExecutableResponse.SegmentResponse(dataSegmentResponse.getId(),
+ dataSegmentResponse.getStatusToDisplay()))
+ .collect(Collectors.toList());
}
//---------------------------------------
@@ -452,8 +449,7 @@ public class JobService extends BasicService implements JobSupporter, ISmartAppl
JobActionEnum.validateValue(action.toUpperCase(Locale.ROOT));
AbstractExecutable job = getManager(NExecutableManager.class, project).getJob(jobId);
- if (SecondStorageUtil.isModelEnable(project, job.getTargetModelId())
- && job.getJobSchedulerMode().equals(JobSchedulerModeEnum.DAG)) {
+ if (job.getJobSchedulerMode().equals(JobSchedulerModeEnum.DAG)) {
checkSegmentState(project, action, job);
}
}
@@ -501,14 +497,12 @@ public class JobService extends BasicService implements JobSupporter, ISmartAppl
jobActionValidate(jobId, project, action);
switch (JobActionEnum.valueOf(action.toUpperCase(Locale.ROOT))) {
case RESUME:
- SecondStorageUtil.checkJobResume(project, jobId);
executableManager.updateJobError(jobId, null, null, null, null);
executableManager.resumeJob(jobId);
UnitOfWork.get().doAfterUnit(afterUnitTask);
MetricsGroup.hostTagCounterInc(MetricsName.JOB_RESUMED, MetricsCategory.PROJECT, project);
break;
case RESTART:
- SecondStorageUtil.checkJobRestart(project, jobId);
executableManager.updateJobError(jobId, null, null, null, null);
executableManager.addFrozenJob(jobId);
executableManager.restartJob(jobId);
@@ -524,7 +518,6 @@ public class JobService extends BasicService implements JobSupporter, ISmartAppl
() -> EventBusFactory.getInstance().postAsync(new JobDiscardNotifier(project, jobType)));
break;
case PAUSE:
- SecondStorageUtil.checkJobPause(project, jobId);
executableManager.pauseJob(jobId);
break;
default:
diff --git a/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java b/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
index e560f1169e..cf56b3bc8c 100644
--- a/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
+++ b/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/JobServiceTest.java
@@ -150,12 +150,12 @@ import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.clickhouse.MockSecondStorage;
+//import io.kyligence.kap.clickhouse.MockSecondStorage;
import org.apache.kylin.engine.spark.job.NSparkCubingJob;
import org.apache.kylin.engine.spark.job.NSparkSnapshotJob;
import org.apache.kylin.engine.spark.job.NTableSamplingJob;
import org.apache.kylin.engine.spark.job.step.NStageForBuild;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.val;
import lombok.var;
@@ -1568,7 +1568,7 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
val model = "89af4ee2-2cdb-4b07-b39e-4c29856309aa";
val project = "default";
- MockSecondStorage.mock("default", new ArrayList<>(), this);
+// MockSecondStorage.mock("default", new ArrayList<>(), this);
val indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), "default");
EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
indexPlanManager.updateIndexPlan(model, indexPlan -> {
@@ -1576,8 +1576,8 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
});
return null;
}, project);
- SecondStorageUtil.initModelMetaData(project, model);
- Assert.assertTrue(SecondStorageUtil.isModelEnable(project, model));
+// SecondStorageUtil.initModelMetaData(project, model);
+// Assert.assertTrue(SecondStorageUtil.isModelEnable(project, model));
val job3 = new DefaultExecutable();
job3.setProject(getProject());
@@ -1600,7 +1600,7 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
val model = "89af4ee2-2cdb-4b07-b39e-4c29856309aa";
val modelMock = RandomUtil.randomUUIDStr();
val project = "default";
- MockSecondStorage.mock("default", new ArrayList<>(), this);
+// MockSecondStorage.mock("default", new ArrayList<>(), this);
val indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), "default");
EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
indexPlanManager.updateIndexPlan(model, indexPlan -> {
@@ -1608,8 +1608,8 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
});
return null;
}, project);
- SecondStorageUtil.initModelMetaData(project, model);
- Assert.assertTrue(SecondStorageUtil.isModelEnable(project, model));
+// SecondStorageUtil.initModelMetaData(project, model);
+// Assert.assertTrue(SecondStorageUtil.isModelEnable(project, model));
val manager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
val job = new DefaultExecutable();
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataModelResponse.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataModelResponse.java
index c0eedd88af..8a1bcc502d 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataModelResponse.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataModelResponse.java
@@ -57,7 +57,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonUnwrapped;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
-import io.kyligence.kap.secondstorage.response.SecondStorageNode;
+//import io.kyligence.kap.secondstorage.response.SecondStorageNode;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.Getter;
@@ -128,8 +128,8 @@ public class NDataModelResponse extends NDataModel {
@JsonProperty("second_storage_size")
private long secondStorageSize;
- @JsonProperty("second_storage_nodes")
- private Map<String, List<SecondStorageNode>> secondStorageNodes;
+// @JsonProperty("second_storage_nodes")
+// private Map<String, List<SecondStorageNode>> secondStorageNodes;
@JsonProperty("second_storage_enabled")
private boolean secondStorageEnabled;
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataSegmentResponse.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataSegmentResponse.java
index 9edd9fbba2..9e6b3811d5 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataSegmentResponse.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/response/NDataSegmentResponse.java
@@ -20,7 +20,7 @@ package org.apache.kylin.rest.response;
import java.io.Serializable;
import java.util.List;
-import java.util.Map;
+//import java.util.Map;
import org.apache.kylin.job.common.SegmentUtil;
import org.apache.kylin.job.execution.AbstractExecutable;
@@ -33,7 +33,7 @@ import org.apache.kylin.metadata.cube.model.NDataflow;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonUnwrapped;
-import io.kyligence.kap.secondstorage.response.SecondStorageNode;
+//import io.kyligence.kap.secondstorage.response.SecondStorageNode;
import lombok.Getter;
import lombok.Setter;
import lombok.val;
@@ -82,8 +82,8 @@ public class NDataSegmentResponse extends NDataSegment {
@JsonProperty("row_count")
private long rowCount;
- @JsonProperty("second_storage_nodes")
- private Map<String, List<SecondStorageNode>> secondStorageNodes;
+// @JsonProperty("second_storage_nodes")
+// private Map<String, List<SecondStorageNode>> secondStorageNodes;
// byte
@JsonProperty("second_storage_size")
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
index c0fbe9c02f..e07ff668b1 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/service/ModelService.java
@@ -102,10 +102,10 @@ import org.apache.calcite.sql.util.SqlBasicVisitor;
import org.apache.calcite.sql.util.SqlVisitor;
import org.apache.calcite.util.Util;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.collections.MapUtils;
+//import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
-import org.apache.kylin.common.KapConfig;
+//import org.apache.kylin.common.KapConfig;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.QueryContext;
import org.apache.kylin.common.event.ModelAddEvent;
@@ -130,17 +130,17 @@ import org.apache.kylin.common.util.RandomUtil;
import org.apache.kylin.common.util.StringHelper;
import org.apache.kylin.common.util.ThreadUtil;
import org.apache.kylin.engine.spark.utils.ComputedColumnEvalUtil;
-import org.apache.kylin.job.SecondStorageJobParamUtil;
+//import org.apache.kylin.job.SecondStorageJobParamUtil;
import org.apache.kylin.job.common.SegmentUtil;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.job.execution.JobTypeEnum;
import org.apache.kylin.job.execution.NExecutableManager;
-import org.apache.kylin.job.handler.SecondStorageIndexCleanJobHandler;
-import org.apache.kylin.job.handler.SecondStorageSegmentCleanJobHandler;
-import org.apache.kylin.job.handler.SecondStorageSegmentLoadJobHandler;
-import org.apache.kylin.job.manager.JobManager;
-import org.apache.kylin.job.model.JobParam;
+//import org.apache.kylin.job.handler.SecondStorageIndexCleanJobHandler;
+//import org.apache.kylin.job.handler.SecondStorageSegmentCleanJobHandler;
+//import org.apache.kylin.job.handler.SecondStorageSegmentLoadJobHandler;
+//import org.apache.kylin.job.manager.JobManager;
+//import org.apache.kylin.job.model.JobParam;
import org.apache.kylin.metadata.acl.AclTCRManager;
import org.apache.kylin.metadata.acl.NDataModelAclParams;
import org.apache.kylin.metadata.cube.cuboid.NAggregationGroup;
@@ -246,7 +246,7 @@ import org.apache.kylin.rest.util.AclPermissionUtil;
import org.apache.kylin.rest.util.ModelTriple;
import org.apache.kylin.rest.util.ModelUtils;
import org.apache.kylin.rest.util.PagingUtil;
-import org.apache.kylin.rest.util.SpringContext;
+//import org.apache.kylin.rest.util.SpringContext;
import org.apache.kylin.source.SourceFactory;
import org.apache.kylin.source.adhocquery.PushDownConverterKeyWords;
import org.apache.kylin.streaming.event.StreamingJobDropEvent;
@@ -274,15 +274,14 @@ import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
import org.apache.kylin.guava30.shaded.common.base.Supplier;
-import io.kyligence.kap.secondstorage.SecondStorage;
-import io.kyligence.kap.secondstorage.SecondStorageNodeHelper;
-import io.kyligence.kap.secondstorage.SecondStorageUpdater;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
-import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
-import io.kyligence.kap.secondstorage.metadata.TableData;
-import io.kyligence.kap.secondstorage.metadata.TablePartition;
-import io.kyligence.kap.secondstorage.response.SecondStorageNode;
-import io.kyligence.kap.secondstorage.util.SecondStorageJobUtil;
+//import io.kyligence.kap.secondstorage.SecondStorage;
+//import io.kyligence.kap.secondstorage.SecondStorageNodeHelper;
+//import io.kyligence.kap.secondstorage.SecondStorageUpdater;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
+//import io.kyligence.kap.secondstorage.metadata.TableData;
+//
+//import io.kyligence.kap.secondstorage.util.SecondStorageJobUtil;
import lombok.Setter;
import lombok.val;
import lombok.var;
@@ -657,7 +656,7 @@ public class ModelService extends AbstractModelService implements TableModelSupp
val pair = getModelsOfCurrentPage(params, tripleList, params.isLite());
models.addAll(pair.getFirst());
// add second storage infos
- ModelUtils.addSecondStorageInfo(project, models);
+// ModelUtils.addSecondStorageInfo(project, models);
filterModels = new DataResult<>(models, pair.getSecond(), offset, limit);
filterModels.setValue(addOldParams(project, filterModels.getValue()));
filterModels.setValue(updateResponseAcl(filterModels.getValue(), project));
@@ -960,8 +959,8 @@ public class ModelService extends AbstractModelService implements TableModelSupp
NDataflow dataflow = dataflowManager.getDataflow(modelId);
List<NDataSegmentResponse> segmentResponseList = getSegmentsResponseCore(modelId, project, start, end, status,
withAllIndexes, withoutAnyIndexes, executables, allToComplement, dataflow);
- addSecondStorageResponse(modelId, project, segmentResponseList, dataflow);
- addSecondStorageDisplayStatus(modelId, project, segmentResponseList);
+// addSecondStorageResponse(modelId, project, segmentResponseList, dataflow);
+// addSecondStorageDisplayStatus(modelId, project, segmentResponseList);
changeSegmentDisplayStatus(modelId, project, segmentResponseList);
segmentResponseList = segmentResponseFilter(statuses, secondStorageStatuses, segmentResponseList, modelId,
project);
@@ -974,8 +973,8 @@ public class ModelService extends AbstractModelService implements TableModelSupp
NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
NDataflow dataflow = dataflowManager.getDataflow(modelId);
List<NDataSegmentResponse> segmentResponseList = getSegmentsResponseCoreByJob(job, dataflow);
- addSecondStorageResponse(modelId, project, segmentResponseList, dataflow);
- addSecondStorageDisplayStatus(modelId, project, segmentResponseList);
+// addSecondStorageResponse(modelId, project, segmentResponseList, dataflow);
+// addSecondStorageDisplayStatus(modelId, project, segmentResponseList);
changeSegmentDisplayStatus(modelId, project, segmentResponseList);
return segmentResponseList;
}
@@ -1035,80 +1034,74 @@ public class ModelService extends AbstractModelService implements TableModelSupp
public void addSecondStorageResponse(String modelId, String project, List<NDataSegmentResponse> segmentResponseList,
NDataflow dataflow) {
- if (!SecondStorageUtil.isModelEnable(project, modelId))
- return;
-
- val tableFlowManager = SecondStorage.tableFlowManager(getConfig(), project);
- val tableFlow = tableFlowManager.get(dataflow.getId()).orElse(null);
- if (tableFlow != null) {
- Map<String, List<TablePartition>> tablePartitions = tableFlow.getTableDataList().stream()
- .flatMap(tableData -> tableData.getPartitions().stream())
- .collect(Collectors.toMap(TablePartition::getSegmentId, partition -> Lists.newArrayList(partition),
- (List<TablePartition> a, List<TablePartition> b) -> {
- a.addAll(b);
- return a;
- }));
-
- List<Set<String>> shards = SecondStorageNodeHelper
- .groupsToShards(SecondStorageUtil.listNodeGroup(getConfig(), project));
-
- segmentResponseList.forEach(segment -> {
- if (tablePartitions.containsKey(segment.getId())) {
- val nodes = new HashSet<String>();
- var partitions = tablePartitions.get(segment.getId());
- for (TablePartition partition : partitions) {
- nodes.addAll(partition.getShardNodes());
- }
-
- segment.setSecondStorageSize(SecondStorageUtil.calculateSecondStorageSize(shards, partitions));
-
- Map<String, List<SecondStorageNode>> pairs = SecondStorageUtil
- .convertNodesToPairs(new ArrayList<>(nodes));
- segment.setSecondStorageNodes(pairs);
- } else {
- segment.setSecondStorageNodes(Collections.emptyMap());
- segment.setSecondStorageSize(0L);
- }
- });
- }
- }
-
- public void addSecondStorageDisplayStatus(String modelId, String project,
- List<NDataSegmentResponse> segmentResponseList) {
- if (!SecondStorageUtil.isModelEnable(project, modelId)) {
- return;
- }
- val jobRunning = getJobRunning(project);
- segmentResponseList.forEach(segmentResponse -> segmentResponse.setStatusSecondStorageToDisplay(
- getSecondStorageSegmentStatus(modelId, project, segmentResponse, jobRunning)));
- }
-
- public SegmentSecondStorageStatusEnum getSecondStorageSegmentStatus(String modelId, String project,
- NDataSegmentResponse segmentResponse, List<AbstractExecutable> jobRunning) {
-
- NExecutableManager manager = getManager(NExecutableManager.class, project);
- boolean hasSecondStorageJobRunning = jobRunning.stream()
- .filter(job -> job.getSegmentIds().contains(segmentResponse.getId()))
- .anyMatch(job -> NExecutableManager.toPO(job, project).getTasks().stream()
- .filter(task -> SecondStorageUtil.EXPORT_STEPS.contains(task.getName()))
- .anyMatch(task -> manager.getOutput(task.getId()).getState().isRunning()));
-
- Set<Long> chSegmentEnableLayouts = SecondStorageUtil.listEnableLayoutBySegment(project, modelId,
- segmentResponse.getId());
- SegmentSecondStorageStatusEnum segmentSecondStorageStatusEnum = null;
-
- if (CollectionUtils.isNotEmpty(chSegmentEnableLayouts)) {
- // segment CH part ready
- segmentSecondStorageStatusEnum = SegmentStatusEnumToDisplay.LOCKED == segmentResponse.getStatusToDisplay()
- ? SegmentSecondStorageStatusEnum.LOCKED
- : SegmentSecondStorageStatusEnum.LOADED;
- }
- if (hasSecondStorageJobRunning) {
- // segment CH part has job running
- segmentSecondStorageStatusEnum = SegmentSecondStorageStatusEnum.LOADING;
- }
- return segmentSecondStorageStatusEnum;
- }
+// if (!SecondStorageUtil.isModelEnable(project, modelId))
+// return;
+//
+// val tableFlowManager = SecondStorage.tableFlowManager(getConfig(), project);
+// val tableFlow = tableFlowManager.get(dataflow.getId()).orElse(null);
+// if (tableFlow != null) {
+// Map<String, List<TablePartition>> tablePartitions = tableFlow.getTableDataList().stream()
+// .flatMap(tableData -> tableData.getPartitions().stream())
+// .collect(Collectors.toMap(TablePartition::getSegmentId, partition -> Lists.newArrayList(partition),
+// (List<TablePartition> a, List<TablePartition> b) -> {
+// a.addAll(b);
+// return a;
+// }));
+//
+// List<Set<String>> shards = SecondStorageNodeHelper
+// .groupsToShards(SecondStorageUtil.listNodeGroup(getConfig(), project));
+//
+// segmentResponseList.forEach(segment -> {
+// if (tablePartitions.containsKey(segment.getId())) {
+// val nodes = new HashSet<String>();
+// var partitions = tablePartitions.get(segment.getId());
+// for (TablePartition partition : partitions) {
+// nodes.addAll(partition.getShardNodes());
+// }
+//
+// segment.setSecondStorageSize(SecondStorageUtil.calculateSecondStorageSize(shards, partitions));
+//
+// Map<String, List<SecondStorageNode>> pairs = SecondStorageUtil
+// .convertNodesToPairs(new ArrayList<>(nodes));
+// segment.setSecondStorageNodes(pairs);
+// } else {
+// segment.setSecondStorageNodes(Collections.emptyMap());
+// segment.setSecondStorageSize(0L);
+// }
+// });
+// }
+ }
+//
+// public void addSecondStorageDisplayStatus(String modelId, String project,
+// List<NDataSegmentResponse> segmentResponseList) {
+// }
+
+// public SegmentSecondStorageStatusEnum getSecondStorageSegmentStatus(String modelId, String project,
+// NDataSegmentResponse segmentResponse, List<AbstractExecutable> jobRunning) {
+//
+// NExecutableManager manager = getManager(NExecutableManager.class, project);
+// boolean hasSecondStorageJobRunning = jobRunning.stream()
+// .filter(job -> job.getSegmentIds().contains(segmentResponse.getId()))
+// .anyMatch(job -> NExecutableManager.toPO(job, project).getTasks().stream()
+// .filter(task -> SecondStorageUtil.EXPORT_STEPS.contains(task.getName()))
+// .anyMatch(task -> manager.getOutput(task.getId()).getState().isRunning()));
+//
+// Set<Long> chSegmentEnableLayouts = SecondStorageUtil.listEnableLayoutBySegment(project, modelId,
+// segmentResponse.getId());
+// SegmentSecondStorageStatusEnum segmentSecondStorageStatusEnum = null;
+//
+// if (CollectionUtils.isNotEmpty(chSegmentEnableLayouts)) {
+// // segment CH part ready
+// segmentSecondStorageStatusEnum = SegmentStatusEnumToDisplay.LOCKED == segmentResponse.getStatusToDisplay()
+// ? SegmentSecondStorageStatusEnum.LOCKED
+// : SegmentSecondStorageStatusEnum.LOADED;
+// }
+// if (hasSecondStorageJobRunning) {
+// // segment CH part has job running
+// segmentSecondStorageStatusEnum = SegmentSecondStorageStatusEnum.LOADING;
+// }
+// return segmentSecondStorageStatusEnum;
+// }
public List<AbstractExecutable> getJobRunning(String project) {
NExecutableManager execManager = getManager(NExecutableManager.class, project);
@@ -1118,28 +1111,28 @@ public class ModelService extends AbstractModelService implements TableModelSupp
public void changeSegmentDisplayStatus(String modelId, String project,
List<NDataSegmentResponse> segmentResponseList) {
- if (!SecondStorageUtil.isModelEnable(project, modelId)) {
- return;
- }
- NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
- NDataflow df = dataflowManager.getDataflow(modelId);
- segmentResponseList.forEach(segmentResponse -> {
- if (SegmentStatusEnumToDisplay.ONLINE == segmentResponse.getStatusToDisplay()) {
- SegmentSecondStorageStatusEnum statusSecondStorageToDisplay = segmentResponse
- .getStatusSecondStorageToDisplay();
- if (MapUtils.isNotEmpty(df.getSegment(segmentResponse.getId()).getLayoutsMap())
- && (SegmentSecondStorageStatusEnum.LOADING == statusSecondStorageToDisplay
- || Objects.isNull(statusSecondStorageToDisplay))) {
- SegmentStatusEnumToDisplay segmentStatusEnumToDisplay = KapConfig.getInstanceFromEnv().isCloud()
- ? SegmentStatusEnumToDisplay.ONLINE_OBJECT_STORAGE
- : SegmentStatusEnumToDisplay.ONLINE_HDFS;
- segmentResponse.setStatusToDisplay(segmentStatusEnumToDisplay);
- } else if (MapUtils.isEmpty(df.getSegment(segmentResponse.getId()).getLayoutsMap())
- && SegmentSecondStorageStatusEnum.LOADED == statusSecondStorageToDisplay) {
- segmentResponse.setStatusToDisplay(SegmentStatusEnumToDisplay.ONLINE_TIERED_STORAGE);
- }
- }
- });
+// if (!SecondStorageUtil.isModelEnable(project, modelId)) {
+// return;
+// }
+// NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
+// NDataflow df = dataflowManager.getDataflow(modelId);
+// segmentResponseList.forEach(segmentResponse -> {
+// if (SegmentStatusEnumToDisplay.ONLINE == segmentResponse.getStatusToDisplay()) {
+// SegmentSecondStorageStatusEnum statusSecondStorageToDisplay = segmentResponse
+// .getStatusSecondStorageToDisplay();
+// if (MapUtils.isNotEmpty(df.getSegment(segmentResponse.getId()).getLayoutsMap())
+// && (SegmentSecondStorageStatusEnum.LOADING == statusSecondStorageToDisplay
+// || Objects.isNull(statusSecondStorageToDisplay))) {
+// SegmentStatusEnumToDisplay segmentStatusEnumToDisplay = KapConfig.getInstanceFromEnv().isCloud()
+// ? SegmentStatusEnumToDisplay.ONLINE_OBJECT_STORAGE
+// : SegmentStatusEnumToDisplay.ONLINE_HDFS;
+// segmentResponse.setStatusToDisplay(segmentStatusEnumToDisplay);
+// } else if (MapUtils.isEmpty(df.getSegment(segmentResponse.getId()).getLayoutsMap())
+// && SegmentSecondStorageStatusEnum.LOADED == statusSecondStorageToDisplay) {
+// segmentResponse.setStatusToDisplay(SegmentStatusEnumToDisplay.ONLINE_TIERED_STORAGE);
+// }
+// }
+// });
}
public List<NDataSegmentResponse> segmentResponseFilter(List<String> statuses, List<String> secondStorageStatuses,
@@ -1147,9 +1140,9 @@ public class ModelService extends AbstractModelService implements TableModelSupp
if (CollectionUtils.isEmpty(statuses) && CollectionUtils.isEmpty(secondStorageStatuses)) {
return segmentResponseList;
}
- if (!SecondStorageUtil.isModelEnable(project, modelId) && CollectionUtils.isNotEmpty(secondStorageStatuses)) {
- secondStorageStatuses.clear();
- }
+// if (!SecondStorageUtil.isModelEnable(project, modelId) && CollectionUtils.isNotEmpty(secondStorageStatuses)) {
+// secondStorageStatuses.clear();
+// }
val statusEnumSet = statuses.stream().map(SegmentStatusEnumToDisplay::getByName).filter(Objects::nonNull)
.collect(Collectors.toSet());
val statusSecondStorageEnum = secondStorageStatuses.stream().map(SegmentSecondStorageStatusEnum::getByName)
@@ -1383,7 +1376,7 @@ public class ModelService extends AbstractModelService implements TableModelSupp
EventBusFactory.getInstance().postSync(new StreamingJobDropEvent(project, modelId));
}
- disableSecondStorageModel(project, modelId);
+// disableSecondStorageModel(project, modelId);
val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
val indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
@@ -1407,7 +1400,7 @@ public class ModelService extends AbstractModelService implements TableModelSupp
NDataSegment[] nDataSegments = segments.toArray(segmentsArray);
nDataflowUpdate.setToRemoveSegs(nDataSegments);
dataflowManager.updateDataflow(nDataflowUpdate);
- disableSecondStorageModel(project, modelId);
+// disableSecondStorageModel(project, modelId);
}
offlineModelIfNecessary(dataflowManager, modelId);
}
@@ -2007,9 +2000,9 @@ public class ModelService extends AbstractModelService implements TableModelSupp
modelRequest.setUuid(model.getUuid());
updateExcludedCheckerResult(project, modelRequest);
// enable second storage
- if (modelRequest.isWithSecondStorage() && !SecondStorageUtil.isModelEnable(project, model.getId())) {
- SecondStorageUtil.initModelMetaData(project, model.getId());
- }
+// if (modelRequest.isWithSecondStorage() && !SecondStorageUtil.isModelEnable(project, model.getId())) {
+// SecondStorageUtil.initModelMetaData(project, model.getId());
+// }
return getManager(NDataModelManager.class, project).getDataModelDesc(model.getUuid());
}, project);
}
@@ -2423,36 +2416,36 @@ public class ModelService extends AbstractModelService implements TableModelSupp
getManager(NIndexPlanManager.class, project).updateIndexPlan(dataflow.getUuid(),
IndexPlan::removeTobeDeleteIndexIfNecessary);
- if (SecondStorageUtil.isModelEnable(project, modelId)) {
- SecondStorage.tableFlowManager(getConfig(), project).get(modelId).ifPresent(tableFlow -> {
- SecondStorageUtil.validateProjectLock(project, Collections.singletonList(LockTypeEnum.LOAD.name()));
-
- val tablePlanManager = SecondStorageUtil.tablePlanManager(getConfig(), project);
- Preconditions.checkState(tablePlanManager.isPresent());
- Preconditions.checkState(tablePlanManager.get().get(modelId).isPresent());
- val tablePlan = tablePlanManager.get().get(modelId).get();
-
- Set<Long> needDeleteLayoutIds = tableFlow.getTableDataList().stream()
- .filter(tableData -> indexIds.contains(tableData.getLayoutID()))
- .filter(tableData -> tableData.getPartitions().stream()
- .allMatch(tablePartition -> segmentIds.contains(tablePartition.getSegmentId())))
- .map(TableData::getLayoutID).filter(layoutId -> dataflow.getIndexPlan()
- .getBaseTableLayoutId().longValue() != layoutId.longValue())
- .collect(Collectors.toSet());
-
- SecondStorageUtil.cleanSegments(project, modelId, new HashSet<>(segmentIds),
- new HashSet<>(indexIds));
-
- tableFlow.update(copied -> copied
- .cleanTableData(tableData -> needDeleteLayoutIds.contains(tableData.getLayoutID())));
- tablePlan.update(t -> t.cleanTable(needDeleteLayoutIds));
-
- val jobHandler = new SecondStorageIndexCleanJobHandler();
- final JobParam param = SecondStorageJobParamUtil.layoutCleanParam(project, modelId,
- BasicService.getUsername(), new HashSet<>(indexIds), new HashSet<>(segmentIds));
- getManager(JobManager.class, project).addJob(param, jobHandler);
- });
- }
+// if (SecondStorageUtil.isModelEnable(project, modelId)) {
+// SecondStorage.tableFlowManager(getConfig(), project).get(modelId).ifPresent(tableFlow -> {
+// SecondStorageUtil.validateProjectLock(project, Collections.singletonList(LockTypeEnum.LOAD.name()));
+//
+// val tablePlanManager = SecondStorageUtil.tablePlanManager(getConfig(), project);
+// Preconditions.checkState(tablePlanManager.isPresent());
+// Preconditions.checkState(tablePlanManager.get().get(modelId).isPresent());
+// val tablePlan = tablePlanManager.get().get(modelId).get();
+//
+// Set<Long> needDeleteLayoutIds = tableFlow.getTableDataList().stream()
+// .filter(tableData -> indexIds.contains(tableData.getLayoutID()))
+// .filter(tableData -> tableData.getPartitions().stream()
+// .allMatch(tablePartition -> segmentIds.contains(tablePartition.getSegmentId())))
+// .map(TableData::getLayoutID).filter(layoutId -> dataflow.getIndexPlan()
+// .getBaseTableLayoutId().longValue() != layoutId.longValue())
+// .collect(Collectors.toSet());
+//
+// SecondStorageUtil.cleanSegments(project, modelId, new HashSet<>(segmentIds),
+// new HashSet<>(indexIds));
+//
+// tableFlow.update(copied -> copied
+// .cleanTableData(tableData -> needDeleteLayoutIds.contains(tableData.getLayoutID())));
+// tablePlan.update(t -> t.cleanTable(needDeleteLayoutIds));
+//
+// val jobHandler = new SecondStorageIndexCleanJobHandler();
+// final JobParam param = SecondStorageJobParamUtil.layoutCleanParam(project, modelId,
+// BasicService.getUsername(), new HashSet<>(indexIds), new HashSet<>(segmentIds));
+// getManager(JobManager.class, project).addJob(param, jobHandler);
+// });
+// }
return null;
}, project);
}
@@ -2465,9 +2458,9 @@ public class ModelService extends AbstractModelService implements TableModelSupp
.filter(NDataModel.NamedColumn::isDimension).collect(Collectors.toList()));
request.setComputedColumnDescs(modelDesc.getComputedColumnDescs());
- if (SecondStorageUtil.isModelEnable(modelDesc.getProject(), modelDesc.getId())) {
- request.setWithSecondStorage(true);
- }
+// if (SecondStorageUtil.isModelEnable(modelDesc.getProject(), modelDesc.getId())) {
+// request.setWithSecondStorage(true);
+// }
return request;
}
@@ -2615,7 +2608,7 @@ public class ModelService extends AbstractModelService implements TableModelSupp
for (NDataSegment existedSegment : segments) {
if (buildSegmentOverlapEnable && NDataModel.ModelType.BATCH == model.getModelType()
&& !model.isMultiPartitionModel() && isBuildAllIndexes && isBuildAllIndexesFinally
- && !SecondStorageUtil.isModelEnable(project, model.getId())) {
+ ) {
isOverlap = existedSegment.getSegRange().overlaps(segmentRangeToBuild)
&& !segmentRangeToBuild.contains(existedSegment.getSegRange());
} else {
@@ -2808,10 +2801,10 @@ public class ModelService extends AbstractModelService implements TableModelSupp
@Transaction(project = 1)
public void deleteSegmentById(String model, String project, String[] ids, boolean force) {
aclEvaluate.checkProjectOperationPermission(project);
- if (SecondStorageUtil.isModelEnable(project, model)) {
- LockTypeEnum.checkLock(LockTypeEnum.LOAD.name(), SecondStorageUtil.getProjectLocks(project));
- }
- SecondStorageUtil.checkSegmentRemove(project, model, ids);
+// if (SecondStorageUtil.isModelEnable(project, model)) {
+// LockTypeEnum.checkLock(LockTypeEnum.LOAD.name(), SecondStorageUtil.getProjectLocks(project));
+// }
+// SecondStorageUtil.checkSegmentRemove(project, model, ids);
NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDesc(model);
if (ManagementType.TABLE_ORIENTED == dataModel.getManagementType()) {
throw new KylinException(PERMISSION_DENIED,
@@ -2831,13 +2824,13 @@ public class ModelService extends AbstractModelService implements TableModelSupp
String.format(Locale.ROOT, MsgPicker.getMsg().getSegNotFound(), id, dataflow.getModelAlias()));
}
}
- if (SecondStorageUtil.isModelEnable(project, model)) {
- SecondStorageUtil.cleanSegments(project, model, idsToDelete);
- val jobHandler = new SecondStorageSegmentCleanJobHandler();
- final JobParam param = SecondStorageJobParamUtil.segmentCleanParam(project, model,
- BasicService.getUsername(), idsToDelete);
- getManager(JobManager.class, project).addJob(param, jobHandler);
- }
+// if (SecondStorageUtil.isModelEnable(project, model)) {
+// SecondStorageUtil.cleanSegments(project, model, idsToDelete);
+// val jobHandler = new SecondStorageSegmentCleanJobHandler();
+// final JobParam param = SecondStorageJobParamUtil.segmentCleanParam(project, model,
+// BasicService.getUsername(), idsToDelete);
+// getManager(JobManager.class, project).addJob(param, jobHandler);
+// }
segmentHelper.removeSegment(project, dataflow.getUuid(), idsToDelete);
offlineModelIfNecessary(dataflowManager, model);
}
@@ -2958,7 +2951,7 @@ public class ModelService extends AbstractModelService implements TableModelSupp
if (SegmentStatusEnum.READY != segment.getStatus() && SegmentStatusEnum.WARNING != segment.getStatus()) {
throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getInvalidMergeSegment());
}
- checkSegmentSecondStorage(modelId, project, segment);
+// checkSegmentSecondStorage(modelId, project, segment);
val segmentStart = segment.getTSRange().getStart();
val segmentEnd = segment.getTSRange().getEnd();
@@ -2973,34 +2966,35 @@ public class ModelService extends AbstractModelService implements TableModelSupp
}
public void checkSegmentSecondStorage(String modelId, String project, NDataSegment segment) {
- if (!SecondStorageUtil.isModelEnable(project, modelId)) {
- return;
- }
- // when enable second storage, need DFS build success, if not, throw exception
- if (MapUtils.isEmpty(segment.getLayoutsMap())) {
- throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getInvalidMergeSegmentWithoutDFS());
- }
+// if (!SecondStorageUtil.isModelEnable(project, modelId)) {
+// return;
+// }
+// // when enable second storage, need DFS build success, if not, throw exception
+// if (MapUtils.isEmpty(segment.getLayoutsMap())) {
+// throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getInvalidMergeSegmentWithoutDFS());
+// }
}
@Transaction(project = 0)
public List<JobInfoResponse.JobInfo> exportSegmentToSecondStorage(String project, String model,
String[] segmentIds) {
- aclEvaluate.checkProjectOperationPermission(project);
- SecondStorageJobUtil.validateSegment(project, model, Arrays.asList(segmentIds));
- checkSegmentsExistById(model, project, segmentIds);
- checkSegmentsStatus(model, project, segmentIds, SegmentStatusEnumToDisplay.LOADING,
- SegmentStatusEnumToDisplay.REFRESHING, SegmentStatusEnumToDisplay.MERGING,
- SegmentStatusEnumToDisplay.LOCKED);
-
- if (!SecondStorage.enabled()) {
- throw new KylinException(JobErrorCode.JOB_CONFIGURATION_ERROR, "!!!No Tiered Storage is installed!!!");
- }
- val jobHandler = new SecondStorageSegmentLoadJobHandler();
-
- final JobParam param = SecondStorageJobParamUtil.of(project, model, BasicService.getUsername(),
- Stream.of(segmentIds));
- return Collections.singletonList(new JobInfoResponse.JobInfo(JobTypeEnum.EXPORT_TO_SECOND_STORAGE.toString(),
- getManager(JobManager.class, project).addJob(param, jobHandler)));
+// aclEvaluate.checkProjectOperationPermission(project);
+// SecondStorageJobUtil.validateSegment(project, model, Arrays.asList(segmentIds));
+// checkSegmentsExistById(model, project, segmentIds);
+// checkSegmentsStatus(model, project, segmentIds, SegmentStatusEnumToDisplay.LOADING,
+// SegmentStatusEnumToDisplay.REFRESHING, SegmentStatusEnumToDisplay.MERGING,
+// SegmentStatusEnumToDisplay.LOCKED);
+//
+// if (!SecondStorage.enabled()) {
+// throw new KylinException(JobErrorCode.JOB_CONFIGURATION_ERROR, "!!!No Tiered Storage is installed!!!");
+// }
+// val jobHandler = new SecondStorageSegmentLoadJobHandler();
+//
+// final JobParam param = SecondStorageJobParamUtil.of(project, model, BasicService.getUsername(),
+// Stream.of(segmentIds));
+// return Collections.singletonList(new JobInfoResponse.JobInfo(JobTypeEnum.EXPORT_TO_SECOND_STORAGE.toString(),
+// getManager(JobManager.class, project).addJob(param, jobHandler)));
+ return null;
}
public BuildBaseIndexResponse updateDataModelSemantic(String project, ModelRequest request) {
@@ -3059,15 +3053,15 @@ public class ModelService extends AbstractModelService implements TableModelSupp
request.getStart(), request.getEnd());
updateExcludedCheckerResult(project, request);
- baseIndexUpdater.setSecondStorageEnabled(request.isWithSecondStorage());
+// baseIndexUpdater.setSecondStorageEnabled(request.isWithSecondStorage());
BuildBaseIndexResponse baseIndexResponse = baseIndexUpdater.update(indexPlanService);
if (!request.isSaveOnly() && (needBuild || baseIndexResponse.hasIndexChange())) {
semanticUpdater.buildForModel(project, modelId);
}
modelChangeSupporters.forEach(listener -> listener.onUpdate(project, modelId));
- changeSecondStorageIfNeeded(project, request, () -> !semanticUpdater.isSignificantChange(originModel,
- modelManager.getDataModelDesc(modelId)));
+// changeSecondStorageIfNeeded(project, request, () -> !semanticUpdater.isSignificantChange(originModel,
+// modelManager.getDataModelDesc(modelId)));
return baseIndexResponse;
}, project);
} catch (TransactionException te) {
@@ -3080,40 +3074,40 @@ public class ModelService extends AbstractModelService implements TableModelSupp
}
public String updateSecondStorageModel(String project, String modelId) {
- if (!SecondStorageUtil.isModelEnable(project, modelId)) {
+// if (!SecondStorageUtil.isModelEnable(project, modelId)) {
return "";
- }
+// }
- SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
- return updater.updateIndex(project, modelId);
+// SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
+// return updater.updateIndex(project, modelId);
}
- private void disableSecondStorageModel(String project, String modelId) {
- if (SecondStorageUtil.isModelEnable(project, modelId)) {
- SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
- updater.disableModel(project, modelId);
- }
- }
+// private void disableSecondStorageModel(String project, String modelId) {
+// if (SecondStorageUtil.isModelEnable(project, modelId)) {
+// SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
+// updater.disableModel(project, modelId);
+// }
+// }
public void changeSecondStorageIfNeeded(String project, ModelRequest request, Supplier<Boolean> needClean) {
// disable second storage
- if (request.getId() != null && SecondStorageUtil.isModelEnable(project, request.getId())
- && !request.isWithSecondStorage()) {
- SecondStorageUtil.validateDisableModel(project, request.getId());
- if (Boolean.TRUE.equals(needClean.get())) {
- disableSecondStorageModel(project, request.getId());
- } else {
- SecondStorageUtil.disableModel(project, request.getId());
- }
- } else if (request.getId() != null && !SecondStorageUtil.isModelEnable(project, request.getId())
- && request.isWithSecondStorage()) {
- val indexPlanManager = getManager(NIndexPlanManager.class, project);
- if (!indexPlanManager.getIndexPlan(request.getId()).containBaseTableLayout()) {
- indexPlanManager.updateIndexPlan(request.getId(), copied -> copied.createAndAddBaseIndex(
- Collections.singletonList(copied.createBaseTableIndex(copied.getModel()))));
- }
- SecondStorageUtil.initModelMetaData(project, request.getId());
- }
+// if (request.getId() != null && SecondStorageUtil.isModelEnable(project, request.getId())
+// && !request.isWithSecondStorage()) {
+// SecondStorageUtil.validateDisableModel(project, request.getId());
+// if (Boolean.TRUE.equals(needClean.get())) {
+// disableSecondStorageModel(project, request.getId());
+// } else {
+// SecondStorageUtil.disableModel(project, request.getId());
+// }
+// } else if (request.getId() != null && !SecondStorageUtil.isModelEnable(project, request.getId())
+// && request.isWithSecondStorage()) {
+// val indexPlanManager = getManager(NIndexPlanManager.class, project);
+// if (!indexPlanManager.getIndexPlan(request.getId()).containBaseTableLayout()) {
+// indexPlanManager.updateIndexPlan(request.getId(), copied -> copied.createAndAddBaseIndex(
+// Collections.singletonList(copied.createBaseTableIndex(copied.getModel()))));
+// }
+// SecondStorageUtil.initModelMetaData(project, request.getId());
+// }
}
public void updateExcludedCheckerResult(String project, ModelRequest request) {
@@ -4243,9 +4237,9 @@ public class ModelService extends AbstractModelService implements TableModelSupp
.filter(col -> ccFullNameSet.contains(col.getAliasDotColumn())) //
.forEach(col -> col.setStatus(NDataModel.ColumnStatus.TOMB));
- if (SecondStorageUtil.isModelEnable(model.getProject(), model.getId())) {
- request.setWithSecondStorage(true);
- }
+// if (SecondStorageUtil.isModelEnable(model.getProject(), model.getId())) {
+// request.setWithSecondStorage(true);
+// }
request.setProject(projectName);
}
diff --git a/src/modeling-service/src/main/java/org/apache/kylin/rest/util/ModelUtils.java b/src/modeling-service/src/main/java/org/apache/kylin/rest/util/ModelUtils.java
index 488c053931..7b311fa1a5 100644
--- a/src/modeling-service/src/main/java/org/apache/kylin/rest/util/ModelUtils.java
+++ b/src/modeling-service/src/main/java/org/apache/kylin/rest/util/ModelUtils.java
@@ -34,11 +34,11 @@ import org.apache.kylin.metadata.model.PartitionDesc;
import org.apache.kylin.metadata.model.NDataModel;
import org.apache.kylin.metadata.model.NDataModelManager;
import org.apache.kylin.rest.constant.ModelAttributeEnum;
-import org.apache.kylin.rest.response.NDataModelResponse;
+//import org.apache.kylin.rest.response.NDataModelResponse;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.val;
public class ModelUtils {
@@ -93,39 +93,39 @@ public class ModelUtils {
Set<ModelAttributeEnum> modelAttributeSet = Sets
.newHashSet(modelAttributes == null ? Collections.emptyList() : modelAttributes);
Set<NDataModel> filteredModels = new HashSet<>();
- if (SecondStorageUtil.isProjectEnable(project)) {
- val secondStorageInfos = SecondStorageUtil.setSecondStorageSizeInfo(models);
- val it = models.listIterator();
- while (it.hasNext()) {
- val secondStorageInfo = secondStorageInfos.get(it.nextIndex());
- NDataModelResponse modelResponse = (NDataModelResponse) it.next();
- modelResponse.setSecondStorageNodes(secondStorageInfo.getSecondStorageNodes());
- modelResponse.setSecondStorageSize(secondStorageInfo.getSecondStorageSize());
- modelResponse.setSecondStorageEnabled(secondStorageInfo.isSecondStorageEnabled());
- }
- if (modelAttributeSet.contains(ModelAttributeEnum.SECOND_STORAGE)) {
- filteredModels.addAll(ModelAttributeEnum.SECOND_STORAGE.filter(models));
- modelAttributeSet.remove(ModelAttributeEnum.SECOND_STORAGE);
- }
- }
+// if (SecondStorageUtil.isProjectEnable(project)) {
+// val secondStorageInfos = SecondStorageUtil.setSecondStorageSizeInfo(models);
+// val it = models.listIterator();
+// while (it.hasNext()) {
+// val secondStorageInfo = secondStorageInfos.get(it.nextIndex());
+// NDataModelResponse modelResponse = (NDataModelResponse) it.next();
+// modelResponse.setSecondStorageNodes(secondStorageInfo.getSecondStorageNodes());
+// modelResponse.setSecondStorageSize(secondStorageInfo.getSecondStorageSize());
+// modelResponse.setSecondStorageEnabled(secondStorageInfo.isSecondStorageEnabled());
+// }
+// if (modelAttributeSet.contains(ModelAttributeEnum.SECOND_STORAGE)) {
+// filteredModels.addAll(ModelAttributeEnum.SECOND_STORAGE.filter(models));
+// modelAttributeSet.remove(ModelAttributeEnum.SECOND_STORAGE);
+// }
+// }
for (val attr : modelAttributeSet) {
filteredModels.addAll(attr.filter(models));
}
return filteredModels;
}
- public static void addSecondStorageInfo(String project, List<NDataModel> models) {
- if (SecondStorageUtil.isProjectEnable(project)) {
- val secondStorageInfos = SecondStorageUtil.setSecondStorageSizeInfo(models);
- val it = models.listIterator();
- while (it.hasNext()) {
- val secondStorageInfo = secondStorageInfos.get(it.nextIndex());
- NDataModelResponse modelResponse = (NDataModelResponse) it.next();
- modelResponse.setSecondStorageNodes(secondStorageInfo.getSecondStorageNodes());
- modelResponse.setSecondStorageSize(secondStorageInfo.getSecondStorageSize());
- modelResponse.setSecondStorageEnabled(secondStorageInfo.isSecondStorageEnabled());
- }
- }
- }
+// public static void addSecondStorageInfo(String project, List<NDataModel> models) {
+// if (SecondStorageUtil.isProjectEnable(project)) {
+// val secondStorageInfos = SecondStorageUtil.setSecondStorageSizeInfo(models);
+// val it = models.listIterator();
+// while (it.hasNext()) {
+// val secondStorageInfo = secondStorageInfos.get(it.nextIndex());
+// NDataModelResponse modelResponse = (NDataModelResponse) it.next();
+// modelResponse.setSecondStorageNodes(secondStorageInfo.getSecondStorageNodes());
+// modelResponse.setSecondStorageSize(secondStorageInfo.getSecondStorageSize());
+// modelResponse.setSecondStorageEnabled(secondStorageInfo.isSecondStorageEnabled());
+// }
+// }
+// }
}
diff --git a/src/modeling-service/src/test/java/org/apache/kylin/rest/util/ModelUtilsTest.java b/src/modeling-service/src/test/java/org/apache/kylin/rest/util/ModelUtilsTest.java
index 79835d54a2..18a8f62557 100644
--- a/src/modeling-service/src/test/java/org/apache/kylin/rest/util/ModelUtilsTest.java
+++ b/src/modeling-service/src/test/java/org/apache/kylin/rest/util/ModelUtilsTest.java
@@ -19,10 +19,10 @@ package org.apache.kylin.rest.util;
import static org.mockito.Mockito.when;
-import java.util.ArrayList;
+//import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.Map;
+//import java.util.Map;
import org.apache.kylin.rest.constant.Constant;
import org.apache.kylin.common.util.NLocalFileMetadataTestCase;
@@ -35,24 +35,24 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PrepareForTest;
+//import org.powermock.api.mockito.PowerMockito;
+//import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.springframework.security.authentication.TestingAuthenticationToken;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
-import org.apache.kylin.guava30.shaded.common.collect.Maps;
+//import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
-import io.kyligence.kap.secondstorage.config.Node;
-import io.kyligence.kap.secondstorage.response.SecondStorageInfo;
-import io.kyligence.kap.secondstorage.response.SecondStorageNode;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.config.Node;
+//import io.kyligence.kap.secondstorage.response.SecondStorageInfo;
+//import io.kyligence.kap.secondstorage.response.SecondStorageNode;
import lombok.val;
@RunWith(PowerMockRunner.class)
-@PrepareForTest({ SecondStorageUtil.class })
+//@PrepareForTest({ SecondStorageUtil.class })
public class ModelUtilsTest extends NLocalFileMetadataTestCase {
private final Authentication authentication = new TestingAuthenticationToken("ADMIN", "ADMIN", Constant.ROLE_ADMIN);
@@ -66,16 +66,16 @@ public class ModelUtilsTest extends NLocalFileMetadataTestCase {
public void teardown() {
}
- private void prepareSecondStorageInfo(List<SecondStorageInfo> secondStorageInfos) {
- val ssi = new SecondStorageInfo().setSecondStorageEnabled(true);
- Map<String, List<SecondStorageNode>> pairs = Maps.newHashMap();
- pairs.put("ssi", Arrays.asList(new SecondStorageNode(new Node())));
- ssi.setSecondStorageSize(1024).setSecondStorageNodes(pairs);
- secondStorageInfos.add(ssi);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "setSecondStorageSizeInfo", List.class))
- .toReturn(secondStorageInfos);
- }
+// private void prepareSecondStorageInfo(List<SecondStorageInfo> secondStorageInfos) {
+// val ssi = new SecondStorageInfo().setSecondStorageEnabled(true);
+// Map<String, List<SecondStorageNode>> pairs = Maps.newHashMap();
+// pairs.put("ssi", Arrays.asList(new SecondStorageNode(new Node())));
+// ssi.setSecondStorageSize(1024).setSecondStorageNodes(pairs);
+// secondStorageInfos.add(ssi);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "setSecondStorageSizeInfo", List.class))
+// .toReturn(secondStorageInfos);
+// }
@Test
public void testComputeExpansionRate() {
@@ -93,20 +93,20 @@ public class ModelUtilsTest extends NLocalFileMetadataTestCase {
Assert.assertTrue(ModelUtils.isArgMatch("ADMIN", false, "ADMIN"));
}
- @Test
- public void testAddSecondStorageInfo() {
- prepareSecondStorageInfo(new ArrayList<>());
- val models = Arrays.asList((NDataModel) new NDataModelResponse());
- ModelUtils.addSecondStorageInfo("default", models);
- val dataModelResp = ((NDataModelResponse) models.get(0));
- Assert.assertEquals(1, dataModelResp.getSecondStorageNodes().size());
- Assert.assertEquals(1024, dataModelResp.getSecondStorageSize());
- Assert.assertEquals(true, dataModelResp.isSecondStorageEnabled());
- }
+// @Test
+// public void testAddSecondStorageInfo() {
+// prepareSecondStorageInfo(new ArrayList<>());
+// val models = Arrays.asList((NDataModel) new NDataModelResponse());
+// ModelUtils.addSecondStorageInfo("default", models);
+// val dataModelResp = ((NDataModelResponse) models.get(0));
+// Assert.assertEquals(1, dataModelResp.getSecondStorageNodes().size());
+// Assert.assertEquals(1024, dataModelResp.getSecondStorageSize());
+// Assert.assertEquals(true, dataModelResp.isSecondStorageEnabled());
+// }
@Test
public void testGetFilterModels() {
- prepareSecondStorageInfo(new ArrayList<>());
+// prepareSecondStorageInfo(new ArrayList<>());
List<NDataModel> mockedModels = Lists.newArrayList();
NDataModelResponse modelSpy4 = Mockito.spy(new NDataModelResponse(new NDataModel()));
when(modelSpy4.isSecondStorageEnabled()).thenReturn(true);
@@ -115,7 +115,7 @@ public class ModelUtilsTest extends NLocalFileMetadataTestCase {
mockedModels);
Assert.assertEquals(1, modelSets.size());
val dataModelResp = ((NDataModelResponse) modelSets.iterator().next());
- Assert.assertEquals(1, dataModelResp.getSecondStorageNodes().size());
+// Assert.assertEquals(1, dataModelResp.getSecondStorageNodes().size());
Assert.assertEquals(1024, dataModelResp.getSecondStorageSize());
Assert.assertEquals(true, dataModelResp.isSecondStorageEnabled());
}
diff --git a/src/query-common/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java b/src/query-common/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
index e2d00d77b3..e45d24158a 100644
--- a/src/query-common/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
+++ b/src/query-common/src/main/java/org/apache/kylin/query/relnode/OLAPContext.java
@@ -39,7 +39,7 @@ import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.QueryContext;
+//import org.apache.kylin.common.QueryContext;
import org.apache.kylin.guava30.shaded.common.base.Preconditions;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
@@ -68,7 +68,7 @@ import org.apache.kylin.storage.StorageContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.val;
@@ -287,8 +287,8 @@ public class OLAPContext {
val streamingRealization = new NativeQueryRealization(modelId, modelAlias,
ctx.storageContext.getStreamingLayoutId(), realizationType, ctx.storageContext.isPartialMatchModel(),
snapshots);
- streamingRealization.setSecondStorage(QueryContext.current().getSecondStorageUsageMap()
- .getOrDefault(streamingRealization.getLayoutId(), false));
+// streamingRealization.setSecondStorage(QueryContext.current().getSecondStorageUsageMap()
+// .getOrDefault(streamingRealization.getLayoutId(), false));
streamingRealization.setStreamingLayout(true);
return streamingRealization;
}
@@ -297,10 +297,10 @@ public class OLAPContext {
String modelId, String modelAlias, List<String> snapshots) {
val realization = new NativeQueryRealization(modelId, modelAlias, ctx.storageContext.getLayoutId(),
realizationType, ctx.storageContext.isPartialMatchModel(), snapshots);
- realization.setSecondStorage(
- QueryContext.current().getSecondStorageUsageMap().getOrDefault(realization.getLayoutId(), false));
- realization.setRecommendSecondStorage(
- recommendSecondStorage(ctx.realization.getProject(), modelId, realizationType));
+// realization.setSecondStorage(
+// QueryContext.current().getSecondStorageUsageMap().getOrDefault(realization.getLayoutId(), false));
+// realization.setRecommendSecondStorage(
+// recommendSecondStorage(ctx.realization.getProject(), modelId, realizationType));
return realization;
}
@@ -308,10 +308,10 @@ public class OLAPContext {
tableSets.addAll(ctx.storageContext.getCandidate().getDerivedTableSnapshots());
}
- private static boolean recommendSecondStorage(String project, String modelId, String realizationType) {
- return QueryMetrics.TABLE_INDEX.equals(realizationType) && SecondStorageUtil.isProjectEnable(project)
- && !SecondStorageUtil.isModelEnable(project, modelId);
- }
+// private static boolean recommendSecondStorage(String project, String modelId, String realizationType) {
+// return QueryMetrics.TABLE_INDEX.equals(realizationType) && SecondStorageUtil.isProjectEnable(project)
+// && !SecondStorageUtil.isModelEnable(project, modelId);
+// }
public static RexInputRef createUniqueInputRefAmongTables(OLAPTableScan table, int columnIdx,
Collection<OLAPTableScan> tables) {
diff --git a/src/query-common/src/main/java/org/apache/kylin/query/routing/Candidate.java b/src/query-common/src/main/java/org/apache/kylin/query/routing/Candidate.java
index 47813efe9a..447996dafc 100644
--- a/src/query-common/src/main/java/org/apache/kylin/query/routing/Candidate.java
+++ b/src/query-common/src/main/java/org/apache/kylin/query/routing/Candidate.java
@@ -24,7 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.commons.collections.CollectionUtils;
+//import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.QueryContext;
import org.apache.kylin.guava30.shaded.common.collect.Maps;
@@ -37,7 +37,7 @@ import org.apache.kylin.metadata.realization.QueryableSeg;
import org.apache.kylin.query.relnode.OLAPContext;
import org.apache.kylin.query.relnode.OLAPContextProp;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.Getter;
import lombok.Setter;
@@ -78,23 +78,23 @@ public class Candidate {
queryableSeg.setStreamingSegments(prunedSegments);
} else {
queryableSeg.setBatchSegments(prunedSegments);
- fillSecondStorageLayouts(df);
+// fillSecondStorageLayouts(df);
}
}
- private void fillSecondStorageLayouts(NDataflow df) {
- Map<String, Set<Long>> secondStorageSegmentLayoutMap = Maps.newHashMap();
- if (SecondStorageUtil.isModelEnable(df.getProject(), df.getId())) {
- for (NDataSegment segment : queryableSeg.getBatchSegments()) {
- Set<Long> chEnableLayoutIds = SecondStorageUtil.listEnableLayoutBySegment(df.getProject(), df.getId(),
- segment.getId());
- if (CollectionUtils.isNotEmpty(chEnableLayoutIds)) {
- secondStorageSegmentLayoutMap.put(segment.getId(), chEnableLayoutIds);
- }
- }
- }
- queryableSeg.setChSegToLayoutsMap(secondStorageSegmentLayoutMap);
- }
+// private void fillSecondStorageLayouts(NDataflow df) {
+// Map<String, Set<Long>> secondStorageSegmentLayoutMap = Maps.newHashMap();
+// if (SecondStorageUtil.isModelEnable(df.getProject(), df.getId())) {
+// for (NDataSegment segment : queryableSeg.getBatchSegments()) {
+// Set<Long> chEnableLayoutIds = SecondStorageUtil.listEnableLayoutBySegment(df.getProject(), df.getId(),
+// segment.getId());
+// if (CollectionUtils.isNotEmpty(chEnableLayoutIds)) {
+// secondStorageSegmentLayoutMap.put(segment.getId(), chEnableLayoutIds);
+// }
+// }
+// }
+// queryableSeg.setChSegToLayoutsMap(secondStorageSegmentLayoutMap);
+// }
public Candidate(IRealization realization, OLAPContext ctx, Map<String, String> matchedJoinsGraphAliasMap) {
this.realization = realization;
diff --git a/src/query-service/src/main/java/org/apache/kylin/rest/service/ModelQueryService.java b/src/query-service/src/main/java/org/apache/kylin/rest/service/ModelQueryService.java
index b7cb028ad3..eb0ef672d4 100644
--- a/src/query-service/src/main/java/org/apache/kylin/rest/service/ModelQueryService.java
+++ b/src/query-service/src/main/java/org/apache/kylin/rest/service/ModelQueryService.java
@@ -50,7 +50,7 @@ import org.springframework.stereotype.Component;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.val;
import lombok.extern.slf4j.Slf4j;
@@ -108,7 +108,7 @@ public class ModelQueryService extends BasicService implements ModelQuerySupport
.collect(Collectors.toList());
if (!modelAttributeSet.isEmpty()) {
- val isProjectEnable = SecondStorageUtil.isProjectEnable(elem.getProjectName());
+ val isProjectEnable = false;
modelTripleList = modelTripleList.parallelStream()
.filter(t -> filterModelAttribute(t, modelAttributeSet, isProjectEnable))
.collect(Collectors.toList());
@@ -138,10 +138,10 @@ public class ModelQueryService extends BasicService implements ModelQuerySupport
private boolean isMatchSecondStorage(ModelTriple modelTriple, boolean isProjectEnable,
Set<ModelAttributeEnum> modelAttributeSet) {
boolean secondStorageMatched = false;
- if (isProjectEnable && modelAttributeSet.contains(ModelAttributeEnum.SECOND_STORAGE)) {
- secondStorageMatched = SecondStorageUtil.isModelEnable(modelTriple.getDataModel().getProject(),
- modelTriple.getDataModel().getId());
- }
+// if (isProjectEnable && modelAttributeSet.contains(ModelAttributeEnum.SECOND_STORAGE)) {
+// secondStorageMatched = SecondStorageUtil.isModelEnable(modelTriple.getDataModel().getProject(),
+// modelTriple.getDataModel().getId());
+// }
return secondStorageMatched;
}
diff --git a/src/query-service/src/main/java/org/apache/kylin/rest/service/QueryHistoryScheduler.java b/src/query-service/src/main/java/org/apache/kylin/rest/service/QueryHistoryScheduler.java
index ad93f112d0..ce445bc518 100644
--- a/src/query-service/src/main/java/org/apache/kylin/rest/service/QueryHistoryScheduler.java
+++ b/src/query-service/src/main/java/org/apache/kylin/rest/service/QueryHistoryScheduler.java
@@ -19,7 +19,7 @@
package org.apache.kylin.rest.service;
import java.util.List;
-import java.util.Map;
+//import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
@@ -39,14 +39,14 @@ import org.apache.kylin.metadata.query.QueryMetrics;
import org.apache.kylin.metadata.query.RDBMSQueryHistoryDAO;
import org.apache.kylin.query.util.SparkJobTrace;
import org.apache.kylin.query.util.SparkJobTraceMetric;
-import org.apache.kylin.rest.util.SpringContext;
+//import org.apache.kylin.rest.util.SpringContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
-import io.kyligence.kap.secondstorage.SecondStorageUpdater;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUpdater;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
public class QueryHistoryScheduler {
@@ -117,7 +117,7 @@ public class QueryHistoryScheduler {
} else {
insertMetrics = metrics;
}
- collectSecondStorageMetric(insertMetrics);
+// collectSecondStorageMetric(insertMetrics);
queryHistoryDAO.insert(insertMetrics);
} catch (Exception th) {
logger.error("Error when write query history", th);
@@ -162,36 +162,36 @@ public class QueryHistoryScheduler {
}
public void collectSecondStorageMetric(List<QueryMetrics> metrics) {
- if (!isSecondStorageQueryMetricCollect) {
- return;
- }
-
- if (!SecondStorageUtil.isGlobalEnable()) {
- return;
- }
-
- SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
-
- for (QueryMetrics metric : metrics) {
- try {
- if (metric.isSecondStorage() && SecondStorageUtil.isProjectEnable(metric.getProjectName())) {
- Map<String, Object> secondStorageMetrics = updater.getQueryMetric(metric.getProjectName(), metric.getQueryId());
-
- if (secondStorageMetrics.containsKey(QueryMetrics.TOTAL_SCAN_BYTES)) {
- metric.setTotalScanBytes((long) secondStorageMetrics.get(QueryMetrics.TOTAL_SCAN_BYTES));
- }
-
- if (secondStorageMetrics.containsKey(QueryMetrics.TOTAL_SCAN_COUNT)) {
- metric.setTotalScanCount((long) secondStorageMetrics.get(QueryMetrics.TOTAL_SCAN_COUNT));
- }
-
- if (secondStorageMetrics.containsKey(QueryMetrics.SOURCE_RESULT_COUNT)) {
- metric.getQueryHistoryInfo().setSourceResultCount((long) secondStorageMetrics.get(QueryMetrics.SOURCE_RESULT_COUNT));
- }
- }
- } catch (Exception e) {
- logger.error("Get tired storage metric fail. query_id: {}, message: {}", metric.getQueryId(), e.getMessage());
- }
- }
+// if (!isSecondStorageQueryMetricCollect) {
+// return;
+// }
+//
+// if (!SecondStorageUtil.isGlobalEnable()) {
+// return;
+// }
+//
+// SecondStorageUpdater updater = SpringContext.getBean(SecondStorageUpdater.class);
+//
+// for (QueryMetrics metric : metrics) {
+// try {
+// if (metric.isSecondStorage() && SecondStorageUtil.isProjectEnable(metric.getProjectName())) {
+// Map<String, Object> secondStorageMetrics = updater.getQueryMetric(metric.getProjectName(), metric.getQueryId());
+//
+// if (secondStorageMetrics.containsKey(QueryMetrics.TOTAL_SCAN_BYTES)) {
+// metric.setTotalScanBytes((long) secondStorageMetrics.get(QueryMetrics.TOTAL_SCAN_BYTES));
+// }
+//
+// if (secondStorageMetrics.containsKey(QueryMetrics.TOTAL_SCAN_COUNT)) {
+// metric.setTotalScanCount((long) secondStorageMetrics.get(QueryMetrics.TOTAL_SCAN_COUNT));
+// }
+//
+// if (secondStorageMetrics.containsKey(QueryMetrics.SOURCE_RESULT_COUNT)) {
+// metric.getQueryHistoryInfo().setSourceResultCount((long) secondStorageMetrics.get(QueryMetrics.SOURCE_RESULT_COUNT));
+// }
+// }
+// } catch (Exception e) {
+// logger.error("Get tired storage metric fail. query_id: {}, message: {}", metric.getQueryId(), e.getMessage());
+// }
+// }
}
}
diff --git a/src/query-service/src/test/java/org/apache/kylin/rest/service/ModelQueryServiceTest.java b/src/query-service/src/test/java/org/apache/kylin/rest/service/ModelQueryServiceTest.java
index fa049383ba..caa616d3dc 100644
--- a/src/query-service/src/test/java/org/apache/kylin/rest/service/ModelQueryServiceTest.java
+++ b/src/query-service/src/test/java/org/apache/kylin/rest/service/ModelQueryServiceTest.java
@@ -46,11 +46,11 @@ import org.springframework.security.core.context.SecurityContextHolder;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import lombok.val;
@RunWith(PowerMockRunner.class)
-@PrepareForTest({ SecondStorageUtil.class, UserGroupInformation.class })
+@PrepareForTest({ UserGroupInformation.class })
public class ModelQueryServiceTest extends NLocalFileMetadataTestCase {
private final Authentication authentication = new TestingAuthenticationToken("ADMIN", "ADMIN", Constant.ROLE_ADMIN);
private ModelQueryService modelQueryService = Mockito.spy(new ModelQueryService());
@@ -79,8 +79,8 @@ public class ModelQueryServiceTest extends NLocalFileMetadataTestCase {
List<ModelAttributeEnum> modelAttributeSet1 = Lists.newArrayList(ModelAttributeEnum.BATCH);
ModelQueryParams modelQueryParams = new ModelQueryParams("", null, true, "default", null, null, "", 0, 10, "",
true, null, modelAttributeSet1, null, null, true, false);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.FALSE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.FALSE);
List<ModelTriple> filteredModels1 = modelQueryService.filterModels(models, modelQueryParams);
Assert.assertEquals(1, filteredModels1.size());
@@ -98,8 +98,8 @@ public class ModelQueryServiceTest extends NLocalFileMetadataTestCase {
ModelAttributeEnum.HYBRID);
ModelQueryParams modelQueryParams = new ModelQueryParams("", null, true, "streaming_test", null, null, "", 0,
10, "", true, null, modelAttributeSet1, null, null, true, false);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.FALSE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.FALSE);
List<ModelTriple> filteredModels1 = modelQueryService.filterModels(models, modelQueryParams);
Assert.assertEquals(2, filteredModels1.size());
@@ -108,29 +108,29 @@ public class ModelQueryServiceTest extends NLocalFileMetadataTestCase {
Assert.assertEquals(0, filteredModels2.size());
}
- @Test
- public void testFilterModelWithSecondStorage() {
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
- PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.TRUE);
-
- val mt1 = new ModelTriple(new NDataflow(), new NDataModelResponse());
- mt1.getDataModel().setModelType(NDataModel.ModelType.BATCH);
-
- List models = Arrays.asList(mt1);
-
- List<ModelAttributeEnum> modelAttributeSet1 = Lists.newArrayList(ModelAttributeEnum.BATCH,
- ModelAttributeEnum.SECOND_STORAGE);
- ModelQueryParams modelQueryParams = new ModelQueryParams("", null, true, "default", null, null, "", 0, 10, "",
- true, null, modelAttributeSet1, null, null, true, false);
- List<ModelTriple> filteredModels1 = modelQueryService.filterModels(models, modelQueryParams);
- Assert.assertEquals(1, filteredModels1.size());
-
- List<ModelAttributeEnum> modelAttributeSet2 = Lists.newArrayList(ModelAttributeEnum.SECOND_STORAGE);
- ModelQueryParams modelQueryParams2 = new ModelQueryParams("", null, true, "default", null, null, "", 0, 10, "",
- true, null, modelAttributeSet2, null, null, true, false);
- List<ModelTriple> filteredModels2 = modelQueryService.filterModels(models, modelQueryParams2);
- Assert.assertEquals(1, filteredModels2.size());
- }
+// @Test
+// public void testFilterModelWithSecondStorage() {
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isProjectEnable")).toReturn(Boolean.TRUE);
+// PowerMockito.stub(PowerMockito.method(SecondStorageUtil.class, "isModelEnable")).toReturn(Boolean.TRUE);
+//
+// val mt1 = new ModelTriple(new NDataflow(), new NDataModelResponse());
+// mt1.getDataModel().setModelType(NDataModel.ModelType.BATCH);
+//
+// List models = Arrays.asList(mt1);
+//
+// List<ModelAttributeEnum> modelAttributeSet1 = Lists.newArrayList(ModelAttributeEnum.BATCH,
+// ModelAttributeEnum.SECOND_STORAGE);
+// ModelQueryParams modelQueryParams = new ModelQueryParams("", null, true, "default", null, null, "", 0, 10, "",
+// true, null, modelAttributeSet1, null, null, true, false);
+// List<ModelTriple> filteredModels1 = modelQueryService.filterModels(models, modelQueryParams);
+// Assert.assertEquals(1, filteredModels1.size());
+//
+// List<ModelAttributeEnum> modelAttributeSet2 = Lists.newArrayList(ModelAttributeEnum.SECOND_STORAGE);
+// ModelQueryParams modelQueryParams2 = new ModelQueryParams("", null, true, "default", null, null, "", 0, 10, "",
+// true, null, modelAttributeSet2, null, null, true, false);
+// List<ModelTriple> filteredModels2 = modelQueryService.filterModels(models, modelQueryParams2);
+// Assert.assertEquals(1, filteredModels2.size());
+// }
@Test
public void testFilterModelAttribute() {
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
deleted file mode 100644
index 8a403e740c..0000000000
--- a/src/second-storage/clickhouse-it/src/test/java/io/kyligence/kap/secondstorage/SecondStorageLockTest.java
+++ /dev/null
@@ -1,3024 +0,0 @@
-/*
- * 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 io.kyligence.kap.secondstorage;
-
-import static io.kyligence.kap.newten.clickhouse.ClickHouseUtils.configClickhouseWith;
-import static io.kyligence.kap.newten.clickhouse.ClickHouseUtils.triggerClickHouseJob;
-import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARAMETER;
-import static org.apache.kylin.common.exception.ServerErrorCode.SECOND_STORAGE_PROJECT_STATUS_ERROR;
-import static org.awaitility.Awaitility.await;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.time.DateUtils;
-import org.apache.kylin.common.ForceToTieredStorage;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.QueryContext;
-import org.apache.kylin.common.SegmentOnlineMode;
-import org.apache.kylin.common.exception.KylinException;
-import org.apache.kylin.common.msg.MsgPicker;
-import org.apache.kylin.common.persistence.metadata.Epoch;
-import org.apache.kylin.common.persistence.transaction.TransactionException;
-import org.apache.kylin.common.persistence.transaction.UnitOfWork;
-import org.apache.kylin.common.util.JsonUtil;
-import org.apache.kylin.common.util.RandomUtil;
-import org.apache.kylin.common.util.Unsafe;
-import org.apache.kylin.engine.spark.IndexDataConstructor;
-import org.apache.kylin.engine.spark.job.NSparkCubingJob;
-import org.apache.kylin.job.SecondStorageCleanJobBuildParams;
-import org.apache.kylin.job.SecondStorageJobParamUtil;
-import org.apache.kylin.job.execution.AbstractExecutable;
-import org.apache.kylin.job.execution.ExecutableState;
-import org.apache.kylin.job.execution.NExecutableManager;
-import org.apache.kylin.job.handler.AbstractJobHandler;
-import org.apache.kylin.job.handler.SecondStorageIndexCleanJobHandler;
-import org.apache.kylin.job.handler.SecondStorageSegmentLoadJobHandler;
-import org.apache.kylin.job.model.JobParam;
-import org.apache.kylin.metadata.cube.model.IndexPlan;
-import org.apache.kylin.metadata.cube.model.LayoutEntity;
-import org.apache.kylin.metadata.cube.model.NDataSegment;
-import org.apache.kylin.metadata.cube.model.NDataflow;
-import org.apache.kylin.metadata.cube.model.NDataflowManager;
-import org.apache.kylin.metadata.cube.model.NIndexPlanManager;
-import org.apache.kylin.metadata.epoch.EpochManager;
-import org.apache.kylin.metadata.epoch.EpochOrchestrator;
-import org.apache.kylin.metadata.model.ColumnDesc;
-import org.apache.kylin.metadata.model.ManagementType;
-import org.apache.kylin.metadata.model.NDataModel;
-import org.apache.kylin.metadata.model.NDataModelManager;
-import org.apache.kylin.metadata.model.PartitionDesc;
-import org.apache.kylin.metadata.model.SegmentRange;
-import org.apache.kylin.metadata.model.SegmentSecondStorageStatusEnum;
-import org.apache.kylin.metadata.model.SegmentStatusEnumToDisplay;
-import org.apache.kylin.metadata.project.EnhancedUnitOfWork;
-import org.apache.kylin.metadata.project.NProjectManager;
-import org.apache.kylin.metadata.query.NativeQueryRealization;
-import org.apache.kylin.metadata.query.QueryHistoryInfo;
-import org.apache.kylin.metadata.query.QueryMetrics;
-import org.apache.kylin.metadata.query.RDBMSQueryHistoryDAO;
-import org.apache.kylin.metadata.query.RDBMSQueryHistoryDaoTest;
-import org.apache.kylin.query.relnode.OLAPContext;
-import org.apache.kylin.rest.controller.NAdminController;
-import org.apache.kylin.rest.controller.NModelController;
-import org.apache.kylin.rest.controller.NQueryController;
-import org.apache.kylin.rest.request.ModelRequest;
-import org.apache.kylin.rest.response.BuildBaseIndexResponse;
-import org.apache.kylin.rest.response.DataResult;
-import org.apache.kylin.rest.response.EnvelopeResponse;
-import org.apache.kylin.rest.response.ExecutableStepResponse;
-import org.apache.kylin.rest.response.JobInfoResponse;
-import org.apache.kylin.rest.response.NDataModelResponse;
-import org.apache.kylin.rest.response.NDataSegmentResponse;
-import org.apache.kylin.rest.response.SimplifiedMeasure;
-import org.apache.kylin.rest.service.AccessService;
-import org.apache.kylin.rest.service.FusionModelService;
-import org.apache.kylin.rest.service.IUserGroupService;
-import org.apache.kylin.rest.service.IndexPlanService;
-import org.apache.kylin.rest.service.JobService;
-import org.apache.kylin.rest.service.ModelBuildService;
-import org.apache.kylin.rest.service.ModelQueryService;
-import org.apache.kylin.rest.service.ModelSemanticHelper;
-import org.apache.kylin.rest.service.ModelService;
-import org.apache.kylin.rest.service.NUserGroupService;
-import org.apache.kylin.rest.service.QueryHistoryScheduler;
-import org.apache.kylin.rest.service.QueryHistoryService;
-import org.apache.kylin.rest.service.SegmentHelper;
-import org.apache.kylin.rest.service.params.MergeSegmentParams;
-import org.apache.kylin.rest.service.params.ModelQueryParams;
-import org.apache.kylin.rest.service.params.RefreshSegmentParams;
-import org.apache.kylin.rest.util.AclEvaluate;
-import org.apache.kylin.rest.util.AclUtil;
-import org.apache.kylin.rest.util.SpringContext;
-import org.apache.kylin.util.ExecAndComp;
-import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.Row;
-import org.apache.spark.sql.SparkSession;
-import org.awaitility.Duration;
-import org.eclipse.jetty.toolchain.test.SimpleRequest;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.RuleChain;
-import org.junit.rules.TestRule;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.stubbing.Answer;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-import org.powermock.modules.junit4.PowerMockRunnerDelegate;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.testcontainers.containers.JdbcDatabaseContainer;
-
-import com.amazonaws.util.EC2MetadataUtils;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.kylin.guava30.shaded.common.base.Preconditions;
-import org.apache.kylin.guava30.shaded.common.collect.ImmutableList;
-import org.apache.kylin.guava30.shaded.common.collect.ImmutableMap;
-import org.apache.kylin.guava30.shaded.common.collect.ImmutableSet;
-import org.apache.kylin.guava30.shaded.common.collect.Lists;
-import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import com.microsoft.azure.storage.StorageException;
-import com.microsoft.azure.storage.blob.CloudBlobClient;
-import com.microsoft.azure.storage.blob.CloudBlobContainer;
-
-import io.kyligence.kap.clickhouse.ClickHouseStorage;
-import io.kyligence.kap.clickhouse.job.AzureBlobClient;
-import io.kyligence.kap.clickhouse.job.BlobUrl;
-import io.kyligence.kap.clickhouse.job.ClickHouse;
-import io.kyligence.kap.clickhouse.job.ClickHouseIndexCleanJob;
-import io.kyligence.kap.clickhouse.job.ClickHouseLoad;
-import io.kyligence.kap.clickhouse.job.ClickHouseModelCleanJob;
-import io.kyligence.kap.clickhouse.job.ClickHouseProjectCleanJob;
-import io.kyligence.kap.clickhouse.job.ClickHouseSegmentCleanJob;
-import io.kyligence.kap.clickhouse.job.ClickhouseLoadFileLoad;
-import io.kyligence.kap.clickhouse.job.DataLoader;
-import io.kyligence.kap.clickhouse.job.LoadContext;
-import io.kyligence.kap.clickhouse.job.S3TableSource;
-import io.kyligence.kap.clickhouse.management.ClickHouseConfigLoader;
-import io.kyligence.kap.clickhouse.parser.ShowDatabasesParser;
-import io.kyligence.kap.newten.clickhouse.ClickHouseSimpleITTestUtils;
-import io.kyligence.kap.newten.clickhouse.ClickHouseUtils;
-import io.kyligence.kap.newten.clickhouse.EmbeddedHttpServer;
-import io.kyligence.kap.secondstorage.config.Node;
-import io.kyligence.kap.secondstorage.database.QueryOperator;
-import io.kyligence.kap.secondstorage.ddl.InsertInto;
-import io.kyligence.kap.secondstorage.ddl.ShowDatabases;
-import io.kyligence.kap.secondstorage.ddl.ShowTables;
-import io.kyligence.kap.secondstorage.ddl.exp.TableIdentifier;
-import io.kyligence.kap.secondstorage.enums.LockOperateTypeEnum;
-import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
-import io.kyligence.kap.secondstorage.factory.SecondStorageFactoryUtils;
-import io.kyligence.kap.secondstorage.management.OpenSecondStorageEndpoint;
-import io.kyligence.kap.secondstorage.management.SecondStorageEndpoint;
-import io.kyligence.kap.secondstorage.management.SecondStorageScheduleService;
-import io.kyligence.kap.secondstorage.management.SecondStorageService;
-import io.kyligence.kap.secondstorage.management.request.ProjectEnableRequest;
-import io.kyligence.kap.secondstorage.management.request.ProjectLoadRequest;
-import io.kyligence.kap.secondstorage.management.request.ProjectLockOperateRequest;
-import io.kyligence.kap.secondstorage.management.request.ProjectNodeRequest;
-import io.kyligence.kap.secondstorage.management.request.ProjectRecoveryResponse;
-import io.kyligence.kap.secondstorage.management.request.ProjectTableSyncResponse;
-import io.kyligence.kap.secondstorage.management.request.RecoverRequest;
-import io.kyligence.kap.secondstorage.management.request.SecondStorageMetadataRequest;
-import io.kyligence.kap.secondstorage.management.request.StorageRequest;
-import io.kyligence.kap.secondstorage.metadata.NodeGroup;
-import io.kyligence.kap.secondstorage.metadata.PartitionType;
-import io.kyligence.kap.secondstorage.metadata.TableData;
-import io.kyligence.kap.secondstorage.metadata.TableEntity;
-import io.kyligence.kap.secondstorage.metadata.TableFlow;
-import io.kyligence.kap.secondstorage.metadata.TablePartition;
-import io.kyligence.kap.secondstorage.metadata.TablePlan;
-import io.kyligence.kap.secondstorage.test.EnableScheduler;
-import io.kyligence.kap.secondstorage.test.EnableTestUser;
-import io.kyligence.kap.secondstorage.test.SharedSparkSession;
-import io.kyligence.kap.secondstorage.test.utils.JobWaiter;
-import lombok.Data;
-import lombok.SneakyThrows;
-import lombok.val;
-import lombok.var;
-
-@RunWith(PowerMockRunner.class)
-@PowerMockRunnerDelegate(JUnit4.class)
-@PowerMockIgnore({ "javax.net.ssl.*", "javax.management.*", "org.apache.hadoop.*", "javax.security.*", "javax.crypto.*",
- "javax.script.*" })
-@PrepareForTest({ SpringContext.class, InsertInto.class, EC2MetadataUtils.class })
-public class SecondStorageLockTest implements JobWaiter {
- private final String modelId = "acfde546-2cc9-4eec-bc92-e3bd46d4e2ee";
- private final String userName = "ADMIN";
- private static final String NULLABLE_STRING = "Nullable(String)";
- private static final String LOW_CARDINALITY_STRING = "LowCardinality(Nullable(String))";
-
- @ClassRule
- public static SharedSparkSession sharedSpark = new SharedSparkSession(ImmutableMap.of("spark.sql.extensions",
- "org.apache.kylin.query.SQLPushDownExtensions", "spark.sql.broadcastTimeout", "900"));
-
- public EnableTestUser enableTestUser = new EnableTestUser();
-
- public EnableScheduler enableScheduler = new EnableScheduler("table_index_incremental",
- "src/test/resources/ut_meta");
-
- @Rule
- public TestRule rule = RuleChain.outerRule(enableTestUser).around(enableScheduler);
-
- @Mock
- private final AclEvaluate aclEvaluate = Mockito.spy(AclEvaluate.class);
- @Mock
- private final JobService jobService = Mockito.spy(JobService.class);
- @Mock
- private final AclUtil aclUtil = Mockito.spy(AclUtil.class);
-
- @InjectMocks
- private SecondStorageService secondStorageService = Mockito.spy(new SecondStorageService());
-
- @InjectMocks
- private ModelService modelService = Mockito.spy(new ModelService());
-
- @Mock
- private SecondStorageEndpoint secondStorageEndpoint = new SecondStorageEndpoint();
-
- @Mock
- private SecondStorageScheduleService secondStorageScheduleService = new SecondStorageScheduleService();
-
- @Mock
- private final IndexPlanService indexPlanService = Mockito.spy(new IndexPlanService());
-
- @Mock
- private final ModelSemanticHelper modelSemanticHelper = Mockito.spy(new ModelSemanticHelper());
-
- @Mock
- private final AccessService accessService = Mockito.spy(AccessService.class);
-
- @Mock
- protected IUserGroupService userGroupService = Mockito.spy(NUserGroupService.class);
-
- @Mock
- private final ModelBuildService modelBuildService = Mockito.spy(ModelBuildService.class);
-
- @Mock
- private final SegmentHelper segmentHelper = Mockito.spy(new SegmentHelper());
-
- @Mock
- private final FusionModelService fusionModelService = Mockito.spy(new FusionModelService());
-
- @Mock
- private final NModelController nModelController = Mockito.spy(new NModelController());
-
- @Mock
- private final ModelQueryService modelQueryService = Mockito.spy(new ModelQueryService());
-
- @Mock
- private final NQueryController nQueryController = Mockito.spy(new NQueryController());
-
- @Mock
- private final QueryHistoryService queryHistoryService = Mockito.spy(new QueryHistoryService());
-
- @Mock
- private final NAdminController nAdminController = Mockito.spy(new NAdminController());
-
- @Mock
- private final OpenSecondStorageEndpoint openSecondStorageEndpoint = Mockito.spy(new OpenSecondStorageEndpoint());
-
- private EmbeddedHttpServer _httpServer = null;
- protected IndexDataConstructor indexDataConstructor;
- private final SparkSession ss = sharedSpark.getSpark();
-
- @Before
- public void setUp() throws Exception {
- PowerMockito.mockStatic(SpringContext.class);
- PowerMockito.when(SpringContext.getBean(SecondStorageUpdater.class))
- .thenAnswer((Answer<SecondStorageUpdater>) invocation -> secondStorageService);
-
- secondStorageEndpoint.setSecondStorageService(secondStorageService);
- secondStorageEndpoint.setModelService(modelService);
-
- secondStorageService.setAclEvaluate(aclEvaluate);
-
- ReflectionTestUtils.setField(aclEvaluate, "aclUtil", aclUtil);
-
- ReflectionTestUtils.setField(modelQueryService, "aclEvaluate", aclEvaluate);
-
- ReflectionTestUtils.setField(indexPlanService, "aclEvaluate", aclEvaluate);
-
- ReflectionTestUtils.setField(modelService, "aclEvaluate", aclEvaluate);
- ReflectionTestUtils.setField(modelService, "accessService", accessService);
- ReflectionTestUtils.setField(modelService, "userGroupService", userGroupService);
- ReflectionTestUtils.setField(modelService, "indexPlanService", indexPlanService);
- ReflectionTestUtils.setField(modelService, "semanticUpdater", modelSemanticHelper);
- ReflectionTestUtils.setField(modelService, "modelBuildService", modelBuildService);
- ReflectionTestUtils.setField(modelService, "modelQuerySupporter", modelQueryService);
-
- ReflectionTestUtils.setField(modelBuildService, "modelService", modelService);
- ReflectionTestUtils.setField(modelBuildService, "segmentHelper", segmentHelper);
- ReflectionTestUtils.setField(modelBuildService, "aclEvaluate", aclEvaluate);
- ReflectionTestUtils.setField(modelBuildService, "accessService", accessService);
- ReflectionTestUtils.setField(modelBuildService, "userGroupService", userGroupService);
-
- ReflectionTestUtils.setField(nModelController, "modelService", modelService);
- ReflectionTestUtils.setField(nModelController, "fusionModelService", fusionModelService);
-
- ReflectionTestUtils.setField(fusionModelService, "modelService", modelService);
-
- ReflectionTestUtils.setField(queryHistoryService, "aclEvaluate", aclEvaluate);
- ReflectionTestUtils.setField(nQueryController, "queryHistoryService", queryHistoryService);
-
- openSecondStorageEndpoint.setModelService(modelService);
- openSecondStorageEndpoint.setSecondStorageEndpoint(secondStorageEndpoint);
- openSecondStorageEndpoint.setSecondStorageService(secondStorageService);
-
- System.setProperty("kylin.job.scheduler.poll-interval-second", "1");
- System.setProperty("kylin.second-storage.class", ClickHouseStorage.class.getCanonicalName());
- System.setProperty("kylin.second-storage.wait-index-build-second", "1");
-
- _httpServer = EmbeddedHttpServer.startServer(getLocalWorkingDirectory());
-
- indexDataConstructor = new IndexDataConstructor(getProject());
- }
-
- @Test
- public void testIsAzure() throws URISyntaxException, StorageException {
- System.setProperty("fs.azure.account.key.devstoreaccount1.localhost:10000",
- "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==");
- System.setProperty("fs.azure.skip.metrics", "true");
- // resolve dns problem
- System.setProperty("fs.azure.storage.emulator.account.name", "devstoreaccount1.localhost:10000");
- System.setProperty("fs.AbstractFileSystem.wasb.impl", "org.apache.hadoop.fs.azure.Wasb");
- System.setProperty("fs.wasb.impl", "org.apache.hadoop.fs.azure.NativeAzureFileSystem");
-
- ClickHouseLoad clickHouseLoad = new ClickHouseLoad();
-
- String v = KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory();
- KylinConfig.getInstanceFromEnv().reloadKylinConfigPropertiesFromSiteProperties();
- System.setProperty("kylin.env.hdfs-working-dir", "abfs://test@devstoreaccount1.localhost:10000/kylin");
- Assert.assertTrue(clickHouseLoad.isAzurePlatform());
- KylinConfig.getInstanceFromEnv().reloadKylinConfigPropertiesFromSiteProperties();
- System.setProperty("kylin.env.hdfs-working-dir", "abfss://test@devstoreaccount1.localhost:10000/kylin");
- Assert.assertTrue(clickHouseLoad.isAzurePlatform());
- KylinConfig.getInstanceFromEnv().reloadKylinConfigPropertiesFromSiteProperties();
- System.setProperty("kylin.env.hdfs-working-dir", "file:///test@devstoreaccount1.localhost:10000/kylin");
- Assert.assertFalse(clickHouseLoad.isAzurePlatform());
- KylinConfig.getInstanceFromEnv().reloadKylinConfigPropertiesFromSiteProperties();
- System.setProperty("kylin.env.hdfs-working-dir", v);
-
- Assert.assertEquals("http", BlobUrl.blob2httpSchema("wasb"));
- Assert.assertEquals("https", BlobUrl.blob2httpSchema("wasbs"));
- Assert.assertEquals("https", BlobUrl.blob2httpSchema("abfs"));
- Assert.assertEquals("https", BlobUrl.blob2httpSchema("abfss"));
-
- CloudBlobClient client = Mockito.mock(CloudBlobClient.class);
- CloudBlobContainer container = Mockito.mock(CloudBlobContainer.class);
- Mockito.when(client.getContainerReference(Mockito.anyString())).thenReturn(container);
- AzureBlobClient azureBlobClient = new AzureBlobClient(client, new BlobUrl());
- Assert.assertThrows(NullPointerException.class, () -> azureBlobClient.generateSasKey("", 1));
- }
-
- @Test
- public void s3TransformFileUrl() {
- PowerMockito.mockStatic(EC2MetadataUtils.class, (t) -> {
- if ("getEC2InstanceRegion".equalsIgnoreCase(t.getMethod().getName())) {
- return "cn-northwest-1";
- }
- return t.callRealMethod();
- });
- String result = new S3TableSource().transformFileUrl(
- "hdfs://liunengdev/kylin_clickhouse/ke_metadata/test/parquet/3070ef88-fa57-4ad6-9a6b-9587cfcd4140/62fea0e8-40ef-4baa-a59f-29822fc17321/20000040001/part-00000-b1920799-ef6f-4c77-b0bf-2e72edb558dd-c000.snappy.parquet",
- "host.docker.internal:9000&test&test123", URI.create("/test"));
- Assert.assertEquals(
- "S3('http://host.docker.internal:9000/liunengdev/kylin_clickhouse/ke_metadata/test/parquet/3070ef88-fa57-4ad6-9a6b-9587cfcd4140/62fea0e8-40ef-4baa-a59f-29822fc17321/20000040001/part-00000-b1920799-ef6f-4c77-b0bf-2e72edb558dd-c000.snappy.parquet', 'test','test123', Parquet)",
- result);
- PowerMockito.mockStatic(EC2MetadataUtils.class, (t) -> {
- if ("getEC2InstanceRegion".equalsIgnoreCase(t.getMethod().getName())) {
- return "northwest-1";
- }
- return t.callRealMethod();
- });
- result = new S3TableSource().transformFileUrl(
- "hdfs://liunengdev/kylin_clickhouse/ke_metadata/test/parquet/3070ef88-fa57-4ad6-9a6b-9587cfcd4140/62fea0e8-40ef-4baa-a59f-29822fc17321/20000040001/part-00000-b1920799-ef6f-4c77-b0bf-2e72edb558dd-c000.snappy.parquet",
- "host.docker.internal:9000&test&test123", URI.create("/test"));
- Assert.assertEquals(
- "S3('http://host.docker.internal:9000/liunengdev/kylin_clickhouse/ke_metadata/test/parquet/3070ef88-fa57-4ad6-9a6b-9587cfcd4140/62fea0e8-40ef-4baa-a59f-29822fc17321/20000040001/part-00000-b1920799-ef6f-4c77-b0bf-2e72edb558dd-c000.snappy.parquet', 'test','test123', Parquet)",
- result);
- }
-
- @Test
- public void testCleanClickhouseTempTable() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val dataflow = dataflowManager.getDataflow(modelId);
- val segs = dataflow.getQueryableSegments().stream().map(NDataSegment::getId)
- .collect(Collectors.toList());
- SecondStorageConcurrentTestUtil.registerWaitPoint(SecondStorageConcurrentTestUtil.WAIT_BEFORE_COMMIT,
- 10000);
- int rows = 0;
- String database = NameUtil.getDatabase(dataflow);
- val jobId2 = triggerClickHouseLoadJob(getProject(), modelId, enableTestUser.getUser(), segs);
- await().pollDelay(Duration.FIVE_SECONDS).until(() -> true);
-
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.pauseJob(jobId2);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID, jobId2);
- waitJobEnd(getProject(), jobId2);
-
- secondStorageScheduleService.secondStorageTempTableCleanTask();
-
- try (Connection connection = DriverManager.getConnection(clickhouse1.getJdbcUrl());
- val stmt = connection.createStatement()) {
- val rs = stmt.executeQuery(String.format(Locale.ROOT,
- "select count() cnt from system.tables where database='%s'", database));
-
- if (rs.next()) {
- rows = rs.getInt("cnt");
- }
- }
- assertNotEquals(0, rows);
- return true;
- });
- }
- }
-
- @Test
- public void testDoubleAcquireLock() throws Exception {
- String modelId = RandomUtil.randomUUIDStr();
- SegmentRange<Long> range = SegmentRange.TimePartitionedSegmentRange.createInfinite();
- SecondStorageLockUtils.acquireLock(modelId, range).lock();
- SecondStorageLockUtils.acquireLock(modelId, range).lock();
- Assert.assertTrue(SecondStorageLockUtils.containsKey(modelId, range));
- }
-
- @Test(expected = ExecutionException.class)
- public void acquireLock() throws Exception {
- System.setProperty("kylin.second-storage.wait-lock-timeout", "60");
- String modelId = RandomUtil.randomUUIDStr();
- SegmentRange<Long> range = SegmentRange.TimePartitionedSegmentRange.createInfinite();
- SegmentRange<Long> range2 = new SegmentRange.TimePartitionedSegmentRange(1L, 2L);
- SecondStorageLockUtils.acquireLock(modelId, range).lock();
- ExecutorService executorService = Executors.newFixedThreadPool(1);
- Future<Boolean> future = executorService.submit(() -> {
- SecondStorageLockUtils.acquireLock(modelId, range2).lock();
- return true;
- });
- future.get();
- executorService.shutdown();
- }
-
- @Test
- public void testBuildAllIndexDeleteLockedIndex() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
- updateIndex("TRANS_ID");
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- buildIncrementalLoadQuery("2012-01-02", "2012-01-03");
- waitAllJobFinish();
-
- long layout = updateIndex("LSTG_SITE_ID");
- val jobs = getNExecutableManager().getAllExecutables().stream().map(AbstractExecutable::getId)
- .collect(Collectors.toSet());
- modelBuildService.addIndexesToSegments(getProject(), modelId,
- Collections.singletonList(getDataFlow().getSegments().getFirstSegment().getId()), null, false,
- 3);
- waitAllJobFinish();
- val existLayouts = getTableFlow().getTableDataList().stream().map(TableData::getLayoutID).distinct()
- .collect(Collectors.toList());
-
- assertEquals(1, existLayouts.size());
- assertEquals(layout, existLayouts.get(0).longValue());
-
- val job = getNExecutableManager().getAllExecutables().stream().filter(e -> !jobs.contains(e.getId()))
- .findFirst();
- assertTrue(job.isPresent());
- assertTrue(NExecutableManager.toPO(job.get(), getProject()).getTasks().stream()
- .anyMatch(e0 -> SecondStorageConstants.STEP_SECOND_STORAGE_INDEX_CLEAN.equals(e0.getName())));
- return true;
- });
- }
- }
-
- @Test
- public void testLoadErrorAndCloseSecondStorage() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
-
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- clickhouse1.stop();
- String jobId = triggerClickHouseLoadJob(getProject(), modelId, userName,
- new ArrayList<>(getAllSegmentIds()));
- waitJobEnd(getProject(), jobId);
- assertEquals(ExecutableState.ERROR, NExecutableManager
- .getInstance(KylinConfig.getInstanceFromEnv(), getProject()).getJob(jobId).getStatus());
- Optional<JobInfoResponse.JobInfo> jobInfo = secondStorageService
- .changeModelSecondStorageState(getProject(), modelId, false);
- jobInfo.ifPresent(job -> waitJobEnd(getProject(), job.getJobId()));
- NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject()).resumeJob(jobId);
- waitJobEnd(getProject(), jobId);
- assertEquals(ExecutableState.SUCCEED, NExecutableManager
- .getInstance(KylinConfig.getInstanceFromEnv(), getProject()).getJob(jobId).getStatus());
- return true;
- });
- }
- }
-
- @Test
- public void testCleanClickhouseDiscardTable() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val dataflow = dataflowManager.getDataflow(modelId);
- val segs = dataflow.getQueryableSegments().stream().map(NDataSegment::getId)
- .collect(Collectors.toList());
- SecondStorageConcurrentTestUtil.registerWaitPoint(SecondStorageConcurrentTestUtil.WAIT_BEFORE_COMMIT,
- 10000);
-
- val jobId1 = triggerClickHouseLoadJob(getProject(), modelId, enableTestUser.getUser(), segs);
- await().pollDelay(Duration.FIVE_SECONDS).until(() -> true);
-
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.pauseJob(jobId1);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID, jobId1);
- waitJobEnd(getProject(), jobId1);
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.discardJob(jobId1);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID);
- waitJobEnd(getProject(), jobId1);
-
- secondStorageScheduleService.secondStorageTempTableCleanTask();
-
- String database = NameUtil.getDatabase(dataflow);
- int rows = 0;
-
- try (Connection connection = DriverManager.getConnection(clickhouse1.getJdbcUrl());
- val stmt = connection.createStatement()) {
- val rs = stmt.executeQuery(String.format(Locale.ROOT,
- "select count() cnt from system.tables where database='%s'", database));
-
- if (rs.next()) {
- rows = rs.getInt("cnt");
-
- }
- }
- assertEquals(0, rows);
- return true;
- });
- }
- }
-
- @Test
- public void testOneNodeIsDownBeforeCommit() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- val jobs = getNExecutableManager().getAllExecutables().stream().map(AbstractExecutable::getId)
- .collect(Collectors.toSet());
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- val buildJob = getNExecutableManager().getAllExecutables().stream()
- .filter(j -> !jobs.contains(j.getId())).findAny().get();
- try {
- SecondStorageUtil.checkJobResume(getProject(), buildJob.getId());
- } catch (Exception ignore) {
- }
-
- assertTrue(SecondStorageUtil.checkMergeFlatTableIsSuccess(buildJob));
- SecondStorageConcurrentTestUtil.registerWaitPoint(SecondStorageConcurrentTestUtil.WAIT_BEFORE_COMMIT,
- 10000);
- val jobId = triggerClickHouseLoadJob(getProject(), modelId, enableTestUser.getUser(),
- getDataFlow().getSegments().stream().map(NDataSegment::getId).collect(Collectors.toList()));
- await().pollDelay(Duration.FIVE_SECONDS).until(() -> true);
- clickhouse1.stop();
- waitJobEnd(getProject(), jobId);
- assertTrue(getNExecutableManager().getAllExecutables().stream()
- .anyMatch(job -> job.getStatus() == ExecutableState.ERROR));
- return true;
- });
- }
- }
-
- @Test
- public void testIncrementBuildLockedLayout() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- wrapWithKylinConfigMode("ANY", "DAG", () -> {
- testIncrementBuildLockedLayout(1, clickhouse1);
- return null;
- });
- }
- }
-
- @Test
- public void testIncrementBuildLockedLayoutChain() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- wrapWithKylinConfigMode("DFS", "CHAIN", () -> {
- testIncrementBuildLockedLayout(1, clickhouse1);
- return null;
- });
- }
- }
-
- private static <T> T wrapWithKylinConfigMode(String onlineMode, String scheduleMode, final Callable<T> lambda)
- throws Exception {
- String defaultOnlineMode = KylinConfig.getInstanceFromEnv().getKylinEngineSegmentOnlineMode();
- String defaultScheduleMode = KylinConfig.getInstanceFromEnv().getJobSchedulerMode();
- System.setProperty("kylin.engine.segment-online-mode", onlineMode);
- System.setProperty("kylin.engine.job-scheduler-mode", scheduleMode);
-
- try {
- return lambda.call();
- } finally {
- System.setProperty("kylin.engine.segment-online-mode", defaultOnlineMode);
- System.setProperty("kylin.engine.job-scheduler-mode", defaultScheduleMode);
- }
- }
-
- @Test
- public void testStatic() {
- JobParam jobParam = new JobParam();
- jobParam.setSecondStorageDeleteLayoutIds(null);
- assertNull(jobParam.getSecondStorageDeleteLayoutIds());
-
- SecondStorageCleanJobBuildParams params = new SecondStorageCleanJobBuildParams(null, jobParam, null);
- params.setSecondStorageDeleteLayoutIds(jobParam.getSecondStorageDeleteLayoutIds());
- assertNull(params.getSecondStorageDeleteLayoutIds());
- }
-
- @Test
- public void testSegmentLoadWithRetry() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- testSegmentLoadWithRetry(1, clickhouse1);
- }
- }
-
- @Test
- public void testModelCleanJobWithAddColumn() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- long jobCnt = getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count();
-
- ModelRequest request = getChangedModelRequest("TRANS_ID");
- EnvelopeResponse<BuildBaseIndexResponse> res1 = nModelController.updateSemantic(request);
- assertEquals("000", res1.getCode());
- jobCnt++;
- assertEquals(jobCnt, getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count());
-
- request = getChangedModelRequest("LEAF_CATEG_ID");
- request.setWithSecondStorage(false);
- EnvelopeResponse<BuildBaseIndexResponse> res2 = nModelController.updateSemantic(request);
- assertEquals("000", res2.getCode());
- jobCnt++;
- assertEquals(jobCnt, getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count());
-
- return null;
- });
- }
- }
-
- @Test
- public void testModelCleanJobWithChangePartition() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- val partitionDesc = getNDataModel().getPartitionDesc();
- partitionDesc.setPartitionDateFormat("yyyy-MM-dd");
-
- long jobCnt = getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count();
-
- getNDataModelManager().updateDataModel(modelId,
- copier -> copier.setManagementType(ManagementType.MODEL_BASED));
-
- ModelRequest request = getChangedModelRequestWithNoPartition("TRANS_ID");
- EnvelopeResponse<BuildBaseIndexResponse> res1 = nModelController.updateSemantic(request);
- assertEquals("000", res1.getCode());
- jobCnt++;
- assertEquals(jobCnt, getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count());
-
- request = getChangedModelRequestWithPartition("LEAF_CATEG_ID", partitionDesc);
- EnvelopeResponse<BuildBaseIndexResponse> res2 = nModelController.updateSemantic(request);
- assertEquals("000", res2.getCode());
- jobCnt++;
- assertEquals(jobCnt, getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count());
-
- request = getChangedModelRequestWithNoPartition("TEST_COUNT_DISTINCT_BITMAP");
- request.setWithSecondStorage(false);
- EnvelopeResponse<BuildBaseIndexResponse> res3 = nModelController.updateSemantic(request);
- assertEquals("000", res3.getCode());
- jobCnt++;
- assertEquals(jobCnt, getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseModelCleanJob.class::isInstance).count());
-
- return null;
- });
- }
- }
-
- @Test
- public void testDropModelWithSecondStorage() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
-
- Assert.assertTrue(getIndexPlan().containBaseTableLayout());
- ModelRequest request = new ModelRequest();
- request.setWithSecondStorage(true);
- request.setUuid(modelId);
- BuildBaseIndexResponse changedResponse = mock(BuildBaseIndexResponse.class);
- Mockito.doCallRealMethod().when(modelService).changeSecondStorageIfNeeded(eq("default"), eq(request),
- eq(() -> true));
- when(changedResponse.hasTableIndexChange()).thenReturn(true);
-
- modelService.dropModel(modelId, getProject());
-
- val tableFlowManager = SecondStorageUtil.tableFlowManager(KylinConfig.getInstanceFromEnv(),
- getProject());
- val tableFlow = tableFlowManager.get().get(modelId);
- Assert.assertFalse(tableFlow.isPresent());
-
- return null;
- });
- }
- }
-
- @Test
- public void testDeleteIndexWithSegmentNotDelete() throws Exception {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val dataflow = dataflowManager.getDataflow(modelId);
- val oldLayouts = dataflow.getFirstSegment().getSegDetails().getLayouts();
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- NIndexPlanManager manager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- manager.updateIndexPlan(modelId, copy -> copy.removeLayouts(Sets.newHashSet(1L), true, true));
- return null;
- }, getProject());
-
- val segment = dataflow.copy().getFirstSegment();
- segment.getSegDetails().setLayouts(oldLayouts);
- val segmentResponse = new NDataSegmentResponse(dataflow, segment);
- assertTrue(segmentResponse.isHasBaseTableIndexData());
- }
-
- @Test
- public void changeSecondStorageIfNeeded() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
-
- Assert.assertTrue(getIndexPlan().containBaseTableLayout());
- ModelRequest request = new ModelRequest();
- request.setWithSecondStorage(false);
- request.setUuid(modelId);
-
- modelService.changeSecondStorageIfNeeded(getProject(), request, () -> true);
- Assert.assertFalse(SecondStorageUtil.isModelEnable(getProject(), modelId));
-
- request.setWithSecondStorage(true);
- modelService.changeSecondStorageIfNeeded(getProject(), request, () -> true);
- Assert.assertTrue(SecondStorageUtil.isModelEnable(getProject(), modelId));
-
- request.setWithSecondStorage(true);
- modelService.changeSecondStorageIfNeeded(getProject(), request, () -> true);
- Assert.assertTrue(SecondStorageUtil.isModelEnable(getProject(), modelId));
-
- request.setWithSecondStorage(false);
- modelService.changeSecondStorageIfNeeded(getProject(), request, () -> true);
- Assert.assertFalse(SecondStorageUtil.isModelEnable(getProject(), modelId));
-
- return null;
- });
- }
- }
-
- @Test
- public void testPurgeModelWithSecondStorage() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
-
- Assert.assertTrue(getIndexPlan().containBaseTableLayout());
-
- modelService.purgeModel(modelId, getProject());
-
- val tableFlowManager = SecondStorageUtil.tableFlowManager(KylinConfig.getInstanceFromEnv(),
- getProject());
- val tableFlow = tableFlowManager.get().get(modelId);
- Assert.assertFalse(tableFlow.isPresent());
- return null;
- });
- }
- }
-
- @Test
- public void testSecondStorageMetricsTwoShardTwoReplica() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse3 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse4 = ClickHouseUtils.startClickHouse()) {
- testSecondStorageMetricsWithRetry(2, clickhouse1, clickhouse2, clickhouse3, clickhouse4);
- }
- }
-
- @Test
- public void testSecondStorageMetricsWithEmpty() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- testSecondStorageMetricsWithEmpty(1, clickhouse1);
- }
- }
-
- @Test
- public void testIncrementalCleanProject() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse.getJdbcUrl(), clickhouse.getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- val request = new ProjectEnableRequest();
- request.setProject(getProject().toUpperCase(Locale.ROOT));
- request.setEnabled(false);
- request.setNewNodes(null);
- val jobInfo = secondStorageEndpoint.enableProjectStorage(request);
- Assert.assertEquals(1, jobInfo.getData().getJobs().size());
- waitAllJobFinish();
- val manager = SecondStorageUtil.tableFlowManager(KylinConfig.getInstanceFromEnv(), getProject());
- val nodeGroupManager = SecondStorageUtil.nodeGroupManager(KylinConfig.getInstanceFromEnv(),
- getProject());
- assertTrue(manager.isPresent());
- assertTrue(nodeGroupManager.isPresent());
- Assert.assertEquals(0, manager.get().listAll().size());
- Assert.assertEquals(0, nodeGroupManager.get().listAll().size());
- return true;
- });
- }
- }
-
- @Test
- public void testSecondStorageMetricsEndpoint() throws Exception {
- EnvelopeResponse<Map<String, Long>> r1 = nQueryController.queryHistoryTiredStorageMetrics(getProject(), "12 3");
- assertEquals(0L, r1.getData().get("total_scan_count").longValue());
- assertEquals(0L, r1.getData().get("total_scan_bytes").longValue());
- assertEquals(0L, r1.getData().get("source_result_count").longValue());
-
- EnvelopeResponse<Map<String, Long>> r2 = nQueryController.queryHistoryTiredStorageMetrics(getProject(), "123");
- assertEquals(0L, r2.getData().get("total_scan_count").longValue());
- assertEquals(0L, r2.getData().get("total_scan_bytes").longValue());
- assertEquals(0L, r2.getData().get("source_result_count").longValue());
-
- val dao = RDBMSQueryHistoryDAO.getInstance();
- QueryMetrics queryMetrics = RDBMSQueryHistoryDaoTest.createQueryMetrics(1580311512000L, 1L, true, getProject(),
- true);
- dao.insert(queryMetrics);
- EnvelopeResponse<Map<String, Long>> r3 = nQueryController.queryHistoryTiredStorageMetrics(getProject(),
- queryMetrics.getQueryId());
- assertEquals(queryMetrics.getTotalScanCount(), r3.getData().get("total_scan_count").longValue());
- assertEquals(queryMetrics.getTotalScanBytes(), r3.getData().get("total_scan_bytes").longValue());
- assertEquals(queryMetrics.getQueryHistoryInfo().getSourceResultCount(),
- r3.getData().get("source_result_count").longValue());
- }
-
- public void testSecondStorageQueryRandom() throws Exception {
- List<Set<String>> shardNames = new ArrayList<>();
- shardNames.add(ImmutableSet.of("node00", "node01"));
- Map<String, AtomicInteger> cnt = new HashMap<>();
-
- IntStream.range(0, 100).forEach(i -> {
- val result = SecondStorageNodeHelper.resolveShardToJDBC(shardNames, QueryContext.current());
- result.forEach(r -> cnt.computeIfAbsent(r, v -> new AtomicInteger()).incrementAndGet());
- });
- assertEquals(2, cnt.keySet().size());
- }
-
- private void testSecondStorageMetricsWithRetry(int replica, JdbcDatabaseContainer<?>... clickhouse)
- throws Exception {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- val r1 = secondStorageService.getQueryMetric(getProject(), "");
- assertEquals(-1L, r1.get(QueryMetrics.TOTAL_SCAN_COUNT));
-
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(),
- true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- val r2 = secondStorageService.getQueryMetric(getProject(), "");
- assertEquals(-1L, r2.get(QueryMetrics.TOTAL_SCAN_COUNT));
-
- Map<String, Map<String, Boolean>> nodeStatusMap = ImmutableMap.of("pair0",
- ImmutableMap.of("node00", false, "node02", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
-
- QueryHistoryScheduler q1 = new QueryHistoryScheduler();
- val rm1 = new QueryMetrics.RealizationMetrics();
- rm1.setSecondStorage(true);
-
- val h1 = new QueryHistoryInfo();
- h1.setRealizationMetrics(ImmutableList.of(rm1));
- val m1 = new QueryMetrics("123");
- m1.setProjectName(getProject());
-
- m1.setQueryHistoryInfo(h1);
-
- System.setProperty("kylin.second-storage.query-metric-collect", "false");
- q1.init();
- q1.collectSecondStorageMetric(ImmutableList.of(m1));
- assertEquals(0L, m1.getTotalScanCount());
- assertEquals(0L, m1.getTotalScanBytes());
- assertEquals(0L, m1.getQueryHistoryInfo().getSourceResultCount());
-
- EnvelopeResponse<String> ers = nAdminController.getPublicConfig();
- assertEquals("000", ers.getCode());
- assertTrue(ers.getData().contains("kylin.second-storage.query-metric-collect"));
-
- System.setProperty("kylin.second-storage.query-metric-collect", "true");
- q1.init();
- q1.collectSecondStorageMetric(ImmutableList.of(m1));
-
- assertEquals(-1L, m1.getTotalScanCount());
- assertEquals(-1L, m1.getTotalScanBytes());
- assertEquals(-1L, m1.getQueryHistoryInfo().getSourceResultCount());
-
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true, "node02", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
-
- // Test
- // Step1: create model and load to second storage
- val layout01 = updateIndex("TRANS_ID");
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02"); // build table index
- checkHttpServer(); // check http server
- triggerClickHouseJob(getDataFlow()); //load into clickhouse
-
- String sql1 = "select TRANS_ID from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-02'";
- assertQueryResult(sql1, layout01);
- String queryId = QueryContext.current().getQueryId();
- QueryHistoryScheduler queryHistoryScheduler = new QueryHistoryScheduler();
- queryHistoryScheduler.init();
- val realizationMetrics = new QueryMetrics.RealizationMetrics();
- realizationMetrics.setSecondStorage(true);
-
- val history = new QueryHistoryInfo();
- history.setRealizationMetrics(ImmutableList.of(realizationMetrics));
- val metric = new QueryMetrics(queryId);
- metric.setProjectName(getProject());
-
- metric.setQueryHistoryInfo(history);
- queryHistoryScheduler.collectSecondStorageMetric(ImmutableList.of(metric));
-
- long scanRows = metric.getTotalScanCount();
- long scanBytes = metric.getTotalScanBytes();
- long resultRows = metric.getQueryHistoryInfo().getSourceResultCount();
-
- String metricSql = "SELECT\n" + " sum(read_rows) AS readRows,\n" + " sum(read_bytes) AS readBytes,\n"
- + " client_name AS clientName,\n" + " sum(result_rows) AS resultRows\n"
- + "FROM system.query_log\n" + "PREWHERE type = 'QueryFinish'\n"
- + "WHERE (type = 'QueryFinish') AND (event_time >= addHours(now(), -1)) AND (event_date >= addDays(now(), -1)) AND (position(client_name, '%s') = 1)\n"
- + "GROUP BY client_name order by client_name desc";
-
- Map<String, Long> chRowsMetric = new HashMap<>(2);
- Map<String, Long> chBytesMetric = new HashMap<>(2);
- Map<String, Long> chResultMetric = new HashMap<>(2);
-
- for (JdbcDatabaseContainer jdbcDatabaseContainer : clickhouse) {
- try (Connection connection = DriverManager.getConnection(jdbcDatabaseContainer.getJdbcUrl());
- val stmt = connection.createStatement()) {
- stmt.execute("SYSTEM FLUSH LOGS");
- val rs = stmt.executeQuery(String.format(Locale.ROOT, metricSql, queryId));
-
- if (rs.next()) {
- long rowsCH = rs.getLong("readRows");
- long bytesCH = rs.getLong("readBytes");
- long resultCH = rs.getLong("resultRows");
- String clientName = rs.getString("clientName");
- chRowsMetric.computeIfAbsent(clientName, k -> 0L);
- chBytesMetric.computeIfAbsent(clientName, k -> 0L);
- chResultMetric.computeIfAbsent(clientName, k -> 0L);
-
- chRowsMetric.computeIfPresent(clientName, (k, v) -> v + rowsCH);
- chBytesMetric.computeIfPresent(clientName, (k, v) -> v + bytesCH);
- chResultMetric.computeIfPresent(clientName, (k, v) -> v + resultCH);
- }
- }
- }
-
- assertEquals(scanRows, chRowsMetric.get(queryId + "_1").longValue());
- assertEquals(scanBytes, chBytesMetric.get(queryId + "_1").longValue());
- assertEquals(resultRows, chResultMetric.get(queryId + "_1").longValue());
-
- clickhouse[0].stop();
- clickhouse[3].stop();
- queryHistoryScheduler.collectSecondStorageMetric(ImmutableList.of(metric));
-
- scanRows = metric.getTotalScanCount();
- scanBytes = metric.getTotalScanBytes();
- resultRows = metric.getQueryHistoryInfo().getSourceResultCount();
- assertEquals(-1L, scanRows);
- assertEquals(-1L, scanBytes);
- assertEquals(-1L, resultRows);
- return true;
- });
- }
-
- private void testSecondStorageMetricsWithEmpty(int replica, JdbcDatabaseContainer<?>... clickhouse)
- throws Exception {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(),
- true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- // Test
- // Step1: create model and load to second storage
- val layout01 = updateIndex("TRANS_ID");
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02"); // build table index
- checkHttpServer(); // check http server
- triggerClickHouseJob(getDataFlow()); //load into clickhouse
-
- String sql1 = "select TRANS_ID from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-02'";
- assertQueryResult(sql1, layout01);
- String queryId = QueryContext.current().getQueryId();
-
- for (JdbcDatabaseContainer jdbcDatabaseContainer : clickhouse) {
- try (Connection connection = DriverManager.getConnection(jdbcDatabaseContainer.getJdbcUrl());
- val stmt = connection.createStatement()) {
- stmt.execute("SYSTEM FLUSH LOGS");
- stmt.execute("truncate table system.query_log");
- }
- }
-
- QueryHistoryScheduler queryHistoryScheduler = new QueryHistoryScheduler();
- queryHistoryScheduler.init();
- val realizationMetrics = new QueryMetrics.RealizationMetrics();
- realizationMetrics.setSecondStorage(true);
-
- val history = new QueryHistoryInfo();
- history.setRealizationMetrics(ImmutableList.of(realizationMetrics));
- val metric = new QueryMetrics(queryId);
- metric.setProjectName(getProject());
-
- metric.setQueryHistoryInfo(history);
- queryHistoryScheduler.collectSecondStorageMetric(ImmutableList.of(metric));
-
- assertEquals(-1, metric.getTotalScanCount());
- assertEquals(-1, metric.getTotalScanBytes());
- return true;
- });
- }
-
- public void testSegmentLoadWithRetry(int replica, JdbcDatabaseContainer<?>... clickhouse) throws Exception {
- PowerMockito.mockStatic(InsertInto.class);
- PowerMockito.when(InsertInto.insertInto(Mockito.anyString(), Mockito.anyString()))
- .thenThrow(new SQLException("broken pipe HTTPSession"));
-
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(),
- true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02"); // build table index
- checkHttpServer(); // check http server
-
- val segments = new HashSet<>(getDataFlow().getSegments());
- AbstractJobHandler localHandler = new SecondStorageSegmentLoadJobHandler();
- JobParam jobParam = SecondStorageJobParamUtil.of(getProject(), getDataFlow().getModel().getUuid(), "ADMIN",
- segments.stream().map(NDataSegment::getId));
- String jobId = ClickHouseUtils.simulateJobMangerAddJob(jobParam, localHandler);
- await().atMost(20, TimeUnit.SECONDS);
- PowerMockito.doCallRealMethod().when(InsertInto.class);
- InsertInto.insertInto(Mockito.anyString(), Mockito.anyString());
- waitJobFinish(getProject(), jobId);
- return true;
- });
- }
-
- @Test
- public void testDeleteShard() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse()) {
-
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- ProjectNodeRequest request = new ProjectNodeRequest();
- request.setProject("wrong");
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, null));
- request.setProject(getProject());
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, null));
-
- List<String> allPairs = SecondStorageNodeHelper.getAllPairs();
- secondStorageService.changeProjectSecondStorageState(getProject(), allPairs, true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- getBuildBaseLayout(new HashSet<>(), new HashSet<>(), clickhouse, replica);
-
- val clickhouseNew = new JdbcDatabaseContainer[] { clickhouse1, clickhouse2 };
- ClickHouseUtils.internalConfigClickHouse(clickhouseNew, replica);
- secondStorageService.changeProjectSecondStorageState(getProject(), ImmutableList.of("pair1"), true);
- assertEquals(clickhouseNew.length, SecondStorageUtil.listProjectNodes(getProject()).size());
-
- checkSegmentDisplay(replica, clickhouse.length / replica);
-
- EnvelopeResponse<ProjectTableSyncResponse> response = secondStorageEndpoint.tableSync(getProject());
- assertEquals("000", response.getCode());
-
- deleteShardParamsCheck(request);
-
- request.setForce(false);
- val shardNames2 = ImmutableList.of("pair0");
- assertThrows(TransactionException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames2));
-
- EnvelopeResponse<List<String>> res1 = this.secondStorageEndpoint.deleteProjectNodes(request,
- ImmutableList.of("pair1"));
- assertEquals("000", res1.getCode());
- assertTrue(res1.getData().isEmpty());
- checkDeletedStatus(Collections.singletonList("pair0"), Collections.singletonList("pair1"));
-
- assertFalse(
- LockTypeEnum.locked(LockTypeEnum.LOAD.name(), SecondStorageUtil.getProjectLocks(getProject())));
-
- assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeProjectSecondStorageState(getProject(), ImmutableList.of("pair1"), true);
- assertEquals(clickhouseNew.length, SecondStorageUtil.listProjectNodes(getProject()).size());
-
- EnvelopeResponse<ProjectTableSyncResponse> response2 = secondStorageEndpoint.tableSync(getProject());
- assertEquals("000", response2.getCode());
-
- secondStorageService.lockOperate(getProject(), Collections.singletonList(LockTypeEnum.LOAD.name()),
- LockOperateTypeEnum.LOCK.name());
-
- request.setForce(true);
- List<String> shardNames = ImmutableList.of("pair0");
- EnvelopeResponse<List<String>> res2 = this.secondStorageEndpoint.deleteProjectNodes(request,
- shardNames);
- assertEquals("000", res2.getCode());
- assertFalse(res2.getData().isEmpty());
- checkDeletedStatus(Collections.singletonList("pair1"), Collections.singletonList("pair0"));
- assertTrue(getTableFlow().getTableDataList().isEmpty());
-
- assertEquals(1,
- getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseProjectCleanJob.class::isInstance)
- .filter(s -> s.getId().equals(res2.getData().get(0))).count());
-
- assertTrue(
- LockTypeEnum.locked(LockTypeEnum.LOAD.name(), SecondStorageUtil.getProjectLocks(getProject())));
- secondStorageService.lockOperate(getProject(), Collections.singletonList(LockTypeEnum.LOAD.name()),
- LockOperateTypeEnum.UNLOCK.name());
-
- return true;
- });
- }
- }
-
- @Test
- public void testMultiProcessLoad() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- buildIncrementalLoadQuery("2012-01-02", "2012-01-03");
- buildIncrementalLoadQuery("2012-01-03", "2012-01-04");
- buildIncrementalLoadQuery("2012-01-04", "2012-01-05");
- buildIncrementalLoadQuery("2012-01-05", "2012-01-06");
- buildIncrementalLoadQuery("2012-01-06", "2012-01-07");
- buildIncrementalLoadQuery("2012-01-07", "2012-01-08");
- buildIncrementalLoadQuery("2012-01-08", "2012-01-09");
- buildIncrementalLoadQuery("2012-01-09", "2012-01-10");
-
- waitAllJobFinish();
- String sql = "select CAL_DT from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-08'";
- OLAPContext.clearThreadLocalContexts();
- QueryContext.current().close();
- QueryContext.current().setRetrySecondStorage(true);
- val result1 = ExecAndComp.queryModel(getProject(), sql).count();
-
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- String jobId = triggerClickHouseLoadJob(getProject(), modelId, "ADMIN",
- getDataFlow().getSegments().stream().map(NDataSegment::getId).collect(Collectors.toList()));
-
- SecondStorageConcurrentTestUtil.registerWaitPoint(SecondStorageConcurrentTestUtil.WAIT_BEFORE_COMMIT,
- 5000);
- clickhouse1.stop();
-
- waitJobEnd(getProject(), jobId);
- assertEquals(ExecutableState.ERROR,
- NExecutableManager.getInstance(getConfig(), getProject()).getJob(jobId).getStatus());
- clickhouse1.start();
- ClickHouseUtils.internalConfigClickHouse(new JdbcDatabaseContainer[] { clickhouse1 }, 1);
-
- SecondStorageConcurrentTestUtil.registerWaitPoint(SecondStorageConcurrentTestUtil.WAIT_PAUSED, 5000);
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.resumeJob(jobId);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID, jobId);
-
- await().atMost(30, TimeUnit.SECONDS)
- .until(() -> NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject())
- .getJob(jobId).getStatus() == ExecutableState.RUNNING);
- await().pollDelay(5, TimeUnit.SECONDS).until(() -> true);
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.pauseJob(jobId);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID, jobId);
- waitJobEnd(getProject(), jobId);
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- val executableManager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(),
- getProject());
- executableManager.resumeJob(jobId);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID, jobId);
- waitAllJobFinish();
-
- OLAPContext.clearThreadLocalContexts();
- QueryContext.current().close();
- QueryContext.current().setRetrySecondStorage(true);
- val result2 = ExecAndComp.queryModel(getProject(), sql).count();
- assertEquals(result1, result2);
- return true;
- });
- }
- }
-
- @Test
- public void testChangeScheduleMode() throws Exception {
- ClickHouseLoad load = new ClickHouseLoad();
- LoadContext context = new LoadContext(null);
- context.deserializeToString("{}");
- List<DataLoader> dataLoaders = new ArrayList<>();
- load.checkResumeFileCorrect(dataLoaders, context);
-
- context.finishSingleFile(new LoadContext.CompletedFileKeyUtil("test", 20000010001L), "file1");
- context.deserializeToString(context.serializeToString());
- load.checkResumeFileCorrect(dataLoaders, context);
- DataLoader dataLoader = mock(DataLoader.class);
- ClickhouseLoadFileLoad fileLoader = Mockito.mock(ClickhouseLoadFileLoad.class);
- Mockito.when(dataLoader.getSingleFileLoaderPerNode())
- .thenReturn(ImmutableMap.of("node", Collections.singletonList(fileLoader)));
- dataLoaders.add(dataLoader);
- assertThrows(IllegalStateException.class, () -> load.checkResumeFileCorrect(dataLoaders, context));
-
- Mockito.when(fileLoader.getParquetFile()).thenReturn("file1");
- load.checkResumeFileCorrect(dataLoaders, context);
- }
-
- @Test
- public void testSizeInNode() throws Exception {
- SecondStorageMetadataRequest request = new SecondStorageMetadataRequest();
- request.setProject("");
- Assert.assertThrows(MsgPicker.getMsg().getEmptyProjectName(), KylinException.class,
- () -> this.secondStorageEndpoint.sizeInNode(request));
- request.setProject("123");
- Assert.assertThrows("123", KylinException.class, () -> this.secondStorageEndpoint.sizeInNode(request));
- request.setProject(getProject());
- Assert.assertThrows(
- String.format(Locale.ROOT, MsgPicker.getMsg().getSecondStorageProjectEnabled(), getProject()),
- KylinException.class, () -> this.secondStorageEndpoint.sizeInNode(request));
-
- Assert.assertThrows(MsgPicker.getMsg().getEmptyProjectName(), KylinException.class,
- () -> this.secondStorageEndpoint.tableSync(""));
- Assert.assertThrows("123", KylinException.class, () -> this.secondStorageEndpoint.tableSync("123"));
- String project = getProject();
- Assert.assertThrows(
- String.format(Locale.ROOT, MsgPicker.getMsg().getSecondStorageProjectEnabled(), getProject()),
- KylinException.class, () -> this.secondStorageEndpoint.tableSync(project));
- }
-
- @Test
- public void testDeleteShardHA() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse3 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse4 = ClickHouseUtils.startClickHouse()) {
-
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1, clickhouse2, clickhouse3, clickhouse4 };
- int replica = 2;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- List<String> allPairs = SecondStorageNodeHelper.getAllPairs();
- secondStorageService.changeProjectSecondStorageState(getProject(), allPairs, true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- getBuildBaseLayout(new HashSet<>(), new HashSet<>(), clickhouse, replica);
- checkSegmentDisplay(replica, clickhouse.length / replica);
-
- ProjectNodeRequest request = new ProjectNodeRequest();
- request.setProject(getProject());
- request.setForce(true);
- EnvelopeResponse<List<String>> res2 = this.secondStorageEndpoint.deleteProjectNodes(request,
- ImmutableList.of("pair1"));
- assertEquals("000", res2.getCode());
- assertFalse(res2.getData().isEmpty());
- checkDeletedStatus(Collections.singletonList("pair0"), Collections.singletonList("pair1"));
- assertTrue(getTableFlow().getTableDataList().isEmpty());
-
- assertFalse(
- LockTypeEnum.locked(LockTypeEnum.LOAD.name(), SecondStorageUtil.getProjectLocks(getProject())));
- assertEquals(1,
- getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseProjectCleanJob.class::isInstance)
- .filter(s -> s.getId().equals(res2.getData().get(0))).count());
- return true;
- });
- }
- }
-
- private void deleteShardParamsCheck(ProjectNodeRequest request) {
- request.setProject(getProject());
- Assert.assertThrows(KylinException.class, () -> this.secondStorageEndpoint.deleteProjectNodes(request, null));
-
- List<String> shardNames1 = Collections.emptyList();
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames1));
-
- List<String> shardNames2 = ImmutableList.of("test");
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames2));
-
- List<String> shardNames3 = ImmutableList.of("pair0", "test");
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames3));
-
- List<String> shardNames4 = ImmutableList.of("pair0", "pair1");
- Assert.assertThrows(KylinException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames4));
-
- List<String> shardNames5 = ImmutableList.of("pair0");
- Assert.assertThrows(TransactionException.class,
- () -> this.secondStorageEndpoint.deleteProjectNodes(request, shardNames5));
- }
-
- private void checkDeletedStatus(List<String> shards, List<String> deletedShards) {
- List<NodeGroup> nodeGroups = getNodeGroups();
- Map<String, List<Node>> clusters = ClickHouseConfigLoader.getInstance().getCluster().getCluster();
- assertEquals(clusters.get(shards.get(0)).size(), nodeGroups.size());
- assertEquals(shards.size(), nodeGroups.get(0).getNodeNames().size());
-
- shards.forEach(
- shard -> nodeGroups
- .forEach(
- nodeGroup -> assertFalse(
- CollectionUtils
- .intersection(clusters.get(shard).stream().map(Node::getName)
- .collect(Collectors.toList()), nodeGroup.getNodeNames())
- .isEmpty())));
-
- deletedShards
- .forEach(
- shard -> nodeGroups
- .forEach(
- nodeGroup -> assertTrue(CollectionUtils
- .intersection(clusters.get(shard).stream().map(Node::getName)
- .collect(Collectors.toList()), nodeGroup.getNodeNames())
- .isEmpty())));
-
- getTableFlow().getTableDataList().forEach(tableData -> tableData.getPartitions().forEach(p -> {
- assertEquals(shards.size(), p.getShardNodes().size());
-
- shards.stream().flatMap(shard -> clusters.get(shard).stream()).map(Node::getName).forEach(nodeName -> {
- assertTrue(p.getSizeInNode().containsKey(nodeName));
- assertTrue(p.getNodeFileMap().containsKey(nodeName));
- });
-
- deletedShards.stream().flatMap(shard -> clusters.get(shard).stream()).map(Node::getName)
- .forEach(nodeName -> {
- assertFalse(p.getSizeInNode().containsKey(nodeName));
- assertFalse(p.getNodeFileMap().containsKey(nodeName));
- });
- }));
- }
-
- @Test
- public void testProjectSecondStorageJobs() {
- try {
- secondStorageEndpoint.getProjectSecondStorageJobs("error");
- } catch (KylinException e) {
- assertEquals(SECOND_STORAGE_PROJECT_STATUS_ERROR.toErrorCode(), e.getErrorCode());
- }
- }
-
- @Test
- public void testProjectLoadWithOneNodeDown() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
-
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- List<String> allPairs = SecondStorageNodeHelper.getAllPairs();
- secondStorageService.changeProjectSecondStorageState(getProject(), allPairs, true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- getBuildBaseLayout(new HashSet<>(), new HashSet<>(), clickhouse, replica);
-
- Map<String, Map<String, Boolean>> nodeStatusMap = ImmutableMap.of("pair0",
- ImmutableMap.of("node00", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
-
- ProjectLoadRequest request = new ProjectLoadRequest();
- request.setProjects(ImmutableList.of(getProject()));
- EnvelopeResponse<List<ProjectRecoveryResponse>> response = this.secondStorageEndpoint
- .projectLoad(request);
- assertEquals("000", response.getCode());
- waitAllJobFinish();
-
- getTableFlow().getTableDataList().forEach(tableData -> tableData.getPartitions()
- .forEach(p -> assertNotEquals(0, (long) p.getSizeInNode().getOrDefault("node00", 0L))));
-
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- return true;
- });
- }
- }
-
- public void testSegmentLoadWithoutRetry() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- testSegmentLoadWithoutRetry(1, clickhouse1);
- }
- }
-
- public void testSegmentLoadWithoutRetry(int replica, JdbcDatabaseContainer<?>... clickhouse) throws Exception {
- PowerMockito.mockStatic(InsertInto.class);
- PowerMockito.when(InsertInto.insertInto(Mockito.anyString(), Mockito.anyString()))
- .thenAnswer((Answer<InsertInto>) invocation -> new InsertInto(TableIdentifier.table("def", "tab")));
-
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(),
- true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02"); // build table index
- checkHttpServer(); // check http server
-
- val segments = new HashSet<>(getDataFlow().getSegments());
- AbstractJobHandler localHandler = new SecondStorageSegmentLoadJobHandler();
- JobParam jobParam = SecondStorageJobParamUtil.of(getProject(), getDataFlow().getModel().getUuid(), "ADMIN",
- segments.stream().map(NDataSegment::getId));
- String jobId = ClickHouseUtils.simulateJobMangerAddJob(jobParam, localHandler);
- await().atMost(20, TimeUnit.SECONDS);
- PowerMockito.doCallRealMethod().when(InsertInto.class);
- InsertInto.insertInto(Mockito.anyString(), Mockito.anyString());
- waitJobEnd(getProject(), jobId);
- return true;
- });
- }
-
- private void testIncrementBuildLockedLayout(int replica, JdbcDatabaseContainer<?>... clickhouse) throws Exception {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- checkSecondStorageLayoutStatus(Collections.emptySet());
-
- val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val dataflow = dataflowManager.getDataflow(modelId);
- String sid1 = dataflow.getQueryableSegments().stream().map(NDataSegment::getId).findAny().get();
-
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
- assertTrue(SecondStorageUtil.listEnableLayoutBySegment(getProject(), modelId, "123").isEmpty());
- assertTrue(SecondStorageUtil.listEnableLayoutBySegment(getProject(), modelId, sid1).isEmpty());
-
- checkSecondStorageLayoutNullWithStatus();
-
- triggerClickHouseJob(dataflow);
- assertFalse(SecondStorageUtil.listEnableLayoutBySegment(getProject(), modelId, sid1).isEmpty());
-
- AtomicInteger indexDeleteJobCnt = new AtomicInteger(0);
- Set<Long> existTablePlanLayoutIds = new HashSet<>();
- Set<Long> existTableDataLayoutIds = new HashSet<>();
- checkSecondStorageLayoutStatus(existTableDataLayoutIds);
-
- // Test
- // Step1: create model and load to second storage
- val layout01 = getBuildBaseLayout(existTablePlanLayoutIds, existTableDataLayoutIds, clickhouse, replica);
-
- {
- List<NDataSegmentResponse> segments = modelService.getSegmentsResponse(modelId, getProject(), "0", "" + (Long.MAX_VALUE - 1), null, null, null, false, null, false, null, null);
- Assert.assertEquals(3, segments.size());
- Assert.assertEquals(SegmentStatusEnumToDisplay.ONLINE, segments.get(0).getStatusToDisplay());
- Assert.assertEquals(SegmentSecondStorageStatusEnum.LOADED, segments.get(0).getStatusSecondStorageToDisplay());
- }
-
- {
- List<NDataSegmentResponse> segments = modelService.getSegmentsResponse(modelId, getProject(), "0",
- "" + (Long.MAX_VALUE - 1), null, null, null, false, null, false, Lists.newArrayList("ONLINE"),
- Lists.newArrayList("LOADED"));
- Assert.assertEquals(3, segments.size());
- Assert.assertEquals(SegmentStatusEnumToDisplay.ONLINE, segments.get(0).getStatusToDisplay());
- Assert.assertEquals(SegmentSecondStorageStatusEnum.LOADED,
- segments.get(0).getStatusSecondStorageToDisplay());
- }
-
- {
- List<NDataSegmentResponse> segments = modelService.getSegmentsResponse(modelId, getProject(), "0",
- "" + (Long.MAX_VALUE - 1), null, null, null, false, null, false, Lists.newArrayList("LOADING"),
- Lists.newArrayList("LOADED"));
- Assert.assertEquals(0, segments.size());
- }
-
- checkSecondStorageLayoutStatus(existTableDataLayoutIds);
-
- // Step2: change model and removed locked index
- val layout02 = updateIndex("LSTG_SITE_ID");
-
- buildSegmentAndLoadCH(layout02); // build new index segment
- existTablePlanLayoutIds.add(layout02);
- existTableDataLayoutIds.add(layout02);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout01);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout02);
- checkSegmentDisplay(replica, clickhouse.length / replica);
-
- // removed old index
- indexPlanService.removeIndexes(getProject(), modelId, ImmutableSet.of(layout01));
- waitAllJobFinish();
- assertEquals(indexDeleteJobCnt.incrementAndGet(), getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseIndexCleanJob.class::isInstance).count());
-
- existTablePlanLayoutIds.remove(layout01);
- existTableDataLayoutIds.remove(layout01);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout02);
-
- // check load second storage button is enable
- assertTrue(checkNDataSegmentResponse());
-
- // Step3: change model and test remove segment of locked index
- val layout03 = updateIndex("IS_EFFECTUAL"); // update model to new
- existTablePlanLayoutIds.add(layout03);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout02);
-
- buildSegmentAndLoadCH(layout03); // build new index segment
- existTableDataLayoutIds.add(layout03);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout02);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout03);
-
- NDataSegment segmentTmp01 = getDataFlow().getSegments().getFirstSegment();
- removeIndexesFromSegments(segmentTmp01.getId(), layout02);
- assertEquals(indexDeleteJobCnt.incrementAndGet(), getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseIndexCleanJob.class::isInstance).count());
-
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(
- getDataFlow().getSegments().stream().map(NDataSegment::getId)
- .filter(segmentId -> !segmentId.equals(segmentTmp01.getId())).collect(Collectors.toSet()),
- layout02);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout03);
-
- // test remove all segment of locked index
- removeIndexesFromSegments(
- getDataFlow().getSegments().stream().map(NDataSegment::getId).collect(Collectors.toList()),
- layout02);
- assertEquals(indexDeleteJobCnt.incrementAndGet(), getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseIndexCleanJob.class::isInstance).count());
- existTablePlanLayoutIds.remove(layout02);
- existTableDataLayoutIds.remove(layout02);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout03);
-
- // Step4: change model and test refresh segment
- long layout04 = testRefreshSegment(existTablePlanLayoutIds, existTableDataLayoutIds, layout03);
- assertEquals(indexDeleteJobCnt.incrementAndGet(), getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseIndexCleanJob.class::isInstance).count());
-
- long layout05 = testCleanSegment(existTablePlanLayoutIds, existTableDataLayoutIds, layout04);
- assertEquals(indexDeleteJobCnt.incrementAndGet(), getNExecutableManager().getAllExecutables().stream()
- .filter(ClickHouseIndexCleanJob.class::isInstance).count());
-
- // test merge
- testMerge(existTablePlanLayoutIds, existTableDataLayoutIds, layout04, layout05);
-
- // Step7: test build new segment and load all to second storage
- Set<String> layout04Segments = getAllSegmentIds();
- buildIncrementalLoadQuery("2012-01-04", "2012-01-05",
- ImmutableSet.of(getIndexPlan().getLayoutEntity(layout05)));
- triggerClickHouseJob(getDataFlow());
-
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(layout04Segments, layout04);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout05);
-
- // Step8: close second storage
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, false);
- return true;
- });
- }
-
- private void checkSecondStorageLayoutStatus(Set<Long> existTableDataLayoutIds) {
- if (!SegmentOnlineMode.ANY.toString().equalsIgnoreCase(NProjectManager.getInstance(getConfig())
- .getProject(getProject()).getConfig().getKylinEngineSegmentOnlineMode())) {
- return;
- }
- Map<Long, Boolean> exist = indexPlanService.getSecondStorageLayoutStatus(getProject(), existTableDataLayoutIds,
- getIndexPlan());
- if (existTableDataLayoutIds.isEmpty()) {
- assertTrue(exist.isEmpty());
- assertFalse(indexPlanService.isSecondStorageLayoutReady(getProject(), modelId, 0));
- }
- for (Long existTableDataLayoutId : existTableDataLayoutIds) {
- assertTrue(indexPlanService.isSecondStorageLayoutReady(getProject(), modelId, existTableDataLayoutId));
- assertTrue(exist.get(existTableDataLayoutId));
- }
- }
-
- private void checkSecondStorageLayoutNullWithStatus() {
- Map<Long, Boolean> exist = indexPlanService.getSecondStorageLayoutStatus(getProject(), ImmutableSet.of(123L),
- getIndexPlan());
- assertTrue(exist.isEmpty());
- }
-
- private long getBuildBaseLayout(Set<Long> existTablePlanLayoutIds, Set<Long> existTableDataLayoutIds,
- JdbcDatabaseContainer<?>[] clickhouse, int replica) throws Exception {
- // Test
- // Step1: create model and load to second storage
- val layout01 = updateIndex("TRANS_ID");
-
- buildIncrementalLoadQuery(); // build table index
- checkHttpServer(); // check http server
- triggerClickHouseJob(getDataFlow()); //load into clickhouse
- existTablePlanLayoutIds.add(layout01);
- existTableDataLayoutIds.add(layout01);
-
- checkSecondStorageBaseMetadata(true, clickhouse.length, replica);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout01);
-
- String sql1 = "select TRANS_ID from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-02' limit 1";
- assertQueryResult(sql1, layout01);
-
- return layout01;
- }
-
- private long testRefreshSegment(Set<Long> existTablePlanLayoutIds, Set<Long> existTableDataLayoutIds,
- long lockedLayoutId) throws IOException, InterruptedException {
- val layout04 = updateIndex("LEAF_CATEG_ID");
- existTablePlanLayoutIds.add(layout04);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
-
- buildSegmentAndLoadCH(layout04); // build new index segment
- existTableDataLayoutIds.add(layout04);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout04);
-
- Set<String> notRefreshedSegments = getAllSegmentIds();
-
- for (NDataSegment segment : getDataFlow().getSegments()) {
- notRefreshedSegments.remove(segment.getId());
- List<JobInfoResponse.JobInfo> jobInfos = modelBuildService.refreshSegmentById(
- new RefreshSegmentParams(getProject(), modelId, new String[] { segment.getId() }));
-
- jobInfos.forEach(j -> waitJobFinish(getProject(), j.getJobId()));
-
- if (notRefreshedSegments.size() == 0) {
- existTablePlanLayoutIds.remove(lockedLayoutId);
- existTableDataLayoutIds.remove(lockedLayoutId);
- }
-
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(notRefreshedSegments, lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout04);
- }
-
- return layout04;
- }
-
- private long testCleanSegment(Set<Long> existTablePlanLayoutIds, Set<Long> existTableDataLayoutIds,
- long lockedLayoutId) throws IOException, InterruptedException {
- // Step5: change model and test clean second storage segment
- val layout05 = updateIndex("TEST_COUNT_DISTINCT_BITMAP");
- existTablePlanLayoutIds.add(layout05);
- buildSegmentAndLoadCH(layout05); // build new index segment
- existTableDataLayoutIds.add(layout05);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout05);
-
- // fix ut coverage, it not work
- // cleanSegments(modelId + "1", null, null); // fix ut coverage, it not work
- getTableFlow().cleanTableData(null); // fix ut coverage, it not work
- getTablePlan().cleanTable(null); // fix ut coverage, it not work
-
- cleanSegments(getAllSegmentIds(), ImmutableSet.of(lockedLayoutId, layout05));
-
- triggerClickHouseJob(getDataFlow());
- existTableDataLayoutIds.add(lockedLayoutId);
- existTableDataLayoutIds.add(layout05);
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layout05);
-
- return layout05;
- }
-
- private void testMerge(Set<Long> existTablePlanLayoutIds, Set<Long> existTableDataLayoutIds, long lockedLayoutId,
- long layoutId) {
- // Step6: test merge
- assertEquals(3, getDataFlow().getSegments().size());
- Collections.sort(getDataFlow().getSegments());
- mergeSegment(ImmutableSet.of(getDataFlow().getSegments().get(0).getId(),
- getDataFlow().getSegments().get(1).getId()));
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layoutId);
-
- EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
- getTableFlow().update(TableFlow::cleanTableData);
- return null;
- }, getProject(), 1, UnitOfWork.DEFAULT_EPOCH_ID);
-
- Collections.sort(getDataFlow().getSegments());
- mergeSegment(ImmutableSet.of(getDataFlow().getSegments().get(0).getId(),
- getDataFlow().getSegments().get(1).getId()));
- checkSecondStorageMetadata(existTablePlanLayoutIds, existTableDataLayoutIds);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), lockedLayoutId);
- checkSecondStorageSegmentMetadata(getAllSegmentIds(), layoutId);
- }
-
- public String getProject() {
- return "table_index_incremental";
- }
-
- private void buildIncrementalLoadQuery() throws Exception {
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- buildIncrementalLoadQuery("2012-01-02", "2012-01-03");
- buildIncrementalLoadQuery("2012-01-03", "2012-01-04");
-
- waitAllJobFinish();
- }
-
- private void buildIncrementalLoadQuery(String start, String end) throws Exception {
- getIndexPlan().getAllLayouts().forEach(layout -> {
- if (!layout.isBaseIndex() || !layout.getIndex().isTableIndex()) {
- indexPlanService.removeIndexes(getProject(), modelId, ImmutableSet.of(layout.getId()));
- }
- });
- buildIncrementalLoadQuery(start, end, new HashSet<>(getIndexPlan().getAllLayouts()));
- }
-
- private void buildIncrementalLoadQuery(String start, String end, Set<LayoutEntity> layoutIds) throws Exception {
- KylinConfig config = KylinConfig.getInstanceFromEnv();
- val dfName = modelId;
- NDataflowManager dsMgr = NDataflowManager.getInstance(config, getProject());
- NDataflow df = dsMgr.getDataflow(dfName);
- val timeRange = new SegmentRange.TimePartitionedSegmentRange(start, end);
- indexDataConstructor.buildIndex(dfName, timeRange, layoutIds, true);
- }
-
- @SneakyThrows
- private void checkHttpServer() throws IOException {
- SimpleRequest sr = new SimpleRequest(_httpServer.serverUri);
- final String content = sr.getString("/");
- assertTrue(content.length() > 0);
- }
-
- private void checkSecondStorageBaseMetadata(boolean isIncremental, int clickhouseNodeSize, int replicaSize) {
- KylinConfig config = KylinConfig.getInstanceFromEnv();
- NDataflowManager dsMgr = NDataflowManager.getInstance(config, getProject());
- NDataflow df = dsMgr.getDataflow(modelId);
- // check TableFlow
- TablePlan plan = SecondStorage.tablePlanManager(config, getProject()).get(modelId).orElse(null);
- Assert.assertNotNull(plan);
- TableFlow flow = SecondStorage.tableFlowManager(config, getProject()).get(modelId).orElse(null);
- Assert.assertNotNull(flow);
-
- Set<LayoutEntity> allLayouts = df.getIndexPlan().getAllLayouts().stream()
- .filter(SecondStorageUtil::isBaseTableIndex).collect(Collectors.toSet());
- Assert.assertEquals(allLayouts.size(), flow.getTableDataList().size());
-
- for (LayoutEntity layoutEntity : allLayouts) {
- TableEntity tableEntity = plan.getEntity(layoutEntity).orElse(null);
- Assert.assertNotNull(tableEntity);
- TableData data = flow.getEntity(layoutEntity).orElse(null);
- Assert.assertNotNull(data);
- Assert.assertEquals(isIncremental ? PartitionType.INCREMENTAL : PartitionType.FULL,
- data.getPartitionType());
- Assert.assertEquals(dsMgr.getDataflow(modelId).getQueryableSegments().size(),
- data.getPartitions().size() / replicaSize);
- TablePartition partition = data.getPartitions().get(0);
- int shards = Math.min(clickhouseNodeSize / replicaSize, tableEntity.getShardNumbers());
- Assert.assertEquals(shards, partition.getShardNodes().size());
- Assert.assertEquals(shards, partition.getSizeInNode().size());
- Assert.assertTrue(partition.getSizeInNode().values().stream().reduce(Long::sum).orElse(0L) > 0L);
- }
- }
-
- private void checkSecondStorageMetadata(Set<Long> tablePlanLayoutIds, Set<Long> tableDataLayoutIds) {
- Set<Long> existTablePlanLayoutId = getTablePlan().getTableMetas().stream().map(TableEntity::getLayoutID)
- .collect(Collectors.toSet());
- assertEquals(existTablePlanLayoutId.size(), tablePlanLayoutIds.size());
- assertTrue(existTablePlanLayoutId.containsAll(tablePlanLayoutIds));
-
- Set<Long> existTableDataLayoutId = getTableFlow().getTableDataList().stream().map(TableData::getLayoutID)
- .collect(Collectors.toSet());
- assertEquals(existTableDataLayoutId.size(), tableDataLayoutIds.size());
- assertTrue(existTableDataLayoutId.containsAll(tableDataLayoutIds));
- }
-
- private void checkSecondStorageSegmentMetadata(Set<String> segmentIds, long layoutId) {
- if (segmentIds.isEmpty()) {
- return;
- }
-
- assertTrue(getTableFlow().getEntity(layoutId).isPresent());
- Set<String> existSegmentIds = getTableFlow().getEntity(layoutId).get().getAllSegments();
- assertEquals(existSegmentIds.size(), segmentIds.size());
- assertTrue(existSegmentIds.containsAll(segmentIds));
- }
-
- private void mergeSegment(Set<String> segmentIds) {
- JobInfoResponse.JobInfo jobInfo = modelBuildService.mergeSegmentsManually(
- new MergeSegmentParams(getProject(), modelId, segmentIds.toArray(new String[] {})));
-
- waitJobFinish(getProject(), jobInfo.getJobId());
- assertTrue(SecondStorageUtil.checkMergeFlatTableIsSuccess(getNExecutableManager().getJob(jobInfo.getJobId())));
-
- try {
- SecondStorageUtil.checkJobRestart(getProject(), jobInfo.getJobId());
- } catch (Exception ignore) {
- }
-
- JobParam chJob = triggerClickHouseJob(getDataFlow());
- waitAllJobFinish(getProject());
- String project = getProject();
- String jobId = chJob.getJobId();
- assertThrows(KylinException.class, () -> SecondStorageUtil.checkJobRestart(project, jobId));
- }
-
- private void checkSegmentDisplay(int replica, int shardCnt) {
- List<NDataSegmentResponse> segments = modelService.getSegmentsResponse(modelId, getProject(), "0",
- "" + (Long.MAX_VALUE - 1), null, null, null, false, null, false, null, null);
- segments.forEach(segment -> {
- assertEquals(shardCnt, segment.getSecondStorageNodes().size());
- assertNotNull(segment.getSecondStorageNodes().values());
- assertTrue(segment.getSecondStorageNodes().values().stream().findFirst().isPresent());
- assertEquals(replica, segment.getSecondStorageNodes().values().stream().findFirst().get().size());
- });
-
- val sum = segments.stream().mapToLong(NDataSegmentResponse::getSecondStorageSize).sum();
-
- ModelQueryParams request = new ModelQueryParams(modelId, null, true, getProject(), null, null, null, 0, 10,
- "last_modify", false, null, null, null, null, true, false);
- DataResult<List<NDataModel>> result = modelService.getModels(request);
-
- result.getValue().stream().filter(nDataModel -> modelId.equals(nDataModel.getId())).forEach(nDataModel -> {
- val nDataModelRes = (NDataModelResponse) nDataModel;
- assertEquals(sum, nDataModelRes.getSecondStorageSize());
- assertEquals(shardCnt, nDataModelRes.getSecondStorageNodes().size());
- assertEquals(replica, nDataModelRes.getSecondStorageNodes().get("pair0").size());
- });
- }
-
- @Data
- public static class BuildBaseIndexUT {
- @JsonProperty("base_table_index")
- public IndexInfo tableIndex;
-
- @Data
- public static class IndexInfo {
- @JsonProperty("layout_id")
- public long layoutId;
- }
- }
-
- private KylinConfig getConfig() {
- return KylinConfig.getInstanceFromEnv();
- }
-
- private TableFlow getTableFlow() {
- Preconditions.checkState(SecondStorageUtil.tableFlowManager(getConfig(), getProject()).isPresent());
- Preconditions.checkState(
- SecondStorageUtil.tableFlowManager(getConfig(), getProject()).get().get(modelId).isPresent());
- return SecondStorageUtil.tableFlowManager(getConfig(), getProject()).get().get(modelId).get();
- }
-
- private TablePlan getTablePlan() {
- Preconditions.checkState(SecondStorageUtil.tablePlanManager(getConfig(), getProject()).isPresent());
- Preconditions.checkState(
- SecondStorageUtil.tablePlanManager(getConfig(), getProject()).get().get(modelId).isPresent());
- return SecondStorageUtil.tablePlanManager(getConfig(), getProject()).get().get(modelId).get();
- }
-
- private IndexPlan getIndexPlan() {
- return NIndexPlanManager.getInstance(getConfig(), getProject()).getIndexPlan(modelId);
- }
-
- private NDataflow getDataFlow() {
- return NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject()).getDataflow(modelId);
- }
-
- private NDataModelManager getNDataModelManager() {
- return NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- }
-
- private NDataModel getNDataModel() {
- return getNDataModelManager().getDataModelDesc(modelId);
- }
-
- private NExecutableManager getNExecutableManager() {
- return NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- }
-
- private List<NodeGroup> getNodeGroups() {
- Preconditions.checkState(SecondStorageUtil.nodeGroupManager(getConfig(), getProject()).isPresent());
- return SecondStorageUtil.nodeGroupManager(getConfig(), getProject()).get().listAll();
- }
-
- private Set<String> getAllSegmentIds() {
- return getDataFlow().getSegments().stream().map(NDataSegment::getId).collect(Collectors.toSet());
- }
-
- private String getSourceUrl() {
- return _httpServer.uriAccessedByDocker.toString();
- }
-
- private static String getLocalWorkingDirectory() {
- String dir = KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory();
- if (dir.startsWith("file://"))
- dir = dir.substring("file://".length());
- try {
- return new File(dir).getCanonicalPath();
- } catch (IOException e) {
- throw new IllegalStateException(e);
- }
- }
-
- private ModelRequest getChangedModelRequest(String columnName) throws IOException {
- KylinConfig.getInstanceFromEnv().setProperty("kylin.metadata.semi-automatic-mode", "true");
-
- var model = getNDataModel();
-
- val request = JsonUtil.readValue(JsonUtil.writeValueAsString(model), ModelRequest.class);
- request.setProject(getProject());
- request.setUuid(modelId);
- request.setAllNamedColumns(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSimplifiedMeasures(model.getAllMeasures().stream().filter(m -> !m.isTomb())
- .map(SimplifiedMeasure::fromMeasure).collect(Collectors.toList()));
- request.setSimplifiedDimensions(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSaveOnly(true);
-
- val columnDesc = model.getRootFactTable().getColumn(columnName).getColumnDesc(); // TRANS_ID
- request.getSimplifiedDimensions().add(getNamedColumn(columnDesc));
- val partitionDesc = model.getPartitionDesc();
- partitionDesc.setPartitionDateFormat("yyyy-MM-dd");
- request.setPartitionDesc(model.getPartitionDesc());
- request.setWithSecondStorage(true);
-
- return JsonUtil.readValue(JsonUtil.writeValueAsString(request), ModelRequest.class);
- }
-
- private ModelRequest getChangedModelRequestWithNoPartition(String columnName) throws IOException {
- KylinConfig.getInstanceFromEnv().setProperty("kylin.metadata.semi-automatic-mode", "true");
-
- var model = getNDataModel();
-
- val request = JsonUtil.readValue(JsonUtil.writeValueAsString(model), ModelRequest.class);
- request.setProject(getProject());
- request.setUuid(modelId);
- request.setAllNamedColumns(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSimplifiedMeasures(model.getAllMeasures().stream().filter(m -> !m.isTomb())
- .map(SimplifiedMeasure::fromMeasure).collect(Collectors.toList()));
- request.setSimplifiedDimensions(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSaveOnly(true);
-
- val columnDesc = model.getRootFactTable().getColumn(columnName).getColumnDesc(); // TRANS_ID
- request.getSimplifiedDimensions().add(getNamedColumn(columnDesc));
- request.setPartitionDesc(null);
- request.setWithSecondStorage(true);
-
- return JsonUtil.readValue(JsonUtil.writeValueAsString(request), ModelRequest.class);
- }
-
- private ModelRequest getChangedModelRequestWithPartition(String columnName, PartitionDesc partitionDesc)
- throws IOException {
- KylinConfig.getInstanceFromEnv().setProperty("kylin.metadata.semi-automatic-mode", "true");
-
- var model = getNDataModel();
-
- val request = JsonUtil.readValue(JsonUtil.writeValueAsString(model), ModelRequest.class);
- request.setProject(getProject());
- request.setUuid(modelId);
- request.setAllNamedColumns(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSimplifiedMeasures(model.getAllMeasures().stream().filter(m -> !m.isTomb())
- .map(SimplifiedMeasure::fromMeasure).collect(Collectors.toList()));
- request.setSimplifiedDimensions(model.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isDimension)
- .collect(Collectors.toList()));
- request.setSaveOnly(true);
-
- val columnDesc = model.getRootFactTable().getColumn(columnName).getColumnDesc(); // TRANS_ID
- request.getSimplifiedDimensions().add(getNamedColumn(columnDesc));
- request.setPartitionDesc(partitionDesc);
- request.setWithSecondStorage(true);
-
- return JsonUtil.readValue(JsonUtil.writeValueAsString(request), ModelRequest.class);
- }
-
- private NDataModel.NamedColumn getNamedColumn(ColumnDesc columnDesc) {
- NDataModel.NamedColumn transIdColumn = new NDataModel.NamedColumn();
- transIdColumn.setId(Integer.parseInt(columnDesc.getId()));
- transIdColumn.setStatus(NDataModel.ColumnStatus.DIMENSION);
- transIdColumn.setName(columnDesc.getTable().getName() + "_" + columnDesc.getName());
- transIdColumn.setAliasDotColumn(columnDesc.getTable().getName() + "." + columnDesc.getName());
- return transIdColumn;
- }
-
- private void buildSegmentAndLoadCH(long layoutId) throws InterruptedException {
- // build new index segment
- val segments = getDataFlow().getSegments().stream()
- .map(s -> new IndexDataConstructor.BuildInfo(modelId, s,
- ImmutableSet.of(getDataFlow().getIndexPlan().getLayoutEntity(layoutId)), true, null))
- .collect(Collectors.toList());
- indexDataConstructor.buildSegments(segments);
- waitAllJobFinish();
- triggerClickHouseJob(getDataFlow());
- }
-
- private long updateIndex(String columnName) throws IOException {
- val indexResponse = modelService.updateDataModelSemantic(getProject(), getChangedModelRequest(columnName));
- val layoutId = JsonUtil.readValue(JsonUtil.writeValueAsString(indexResponse),
- BuildBaseIndexUT.class).tableIndex.layoutId;
-
- getNExecutableManager().getAllExecutables().forEach(exec -> waitJobFinish(getProject(), exec.getId()));
- return layoutId;
- }
-
- private boolean checkNDataSegmentResponse() {
- NDataSegmentResponse res = new NDataSegmentResponse(getDataFlow(), getDataFlow().getFirstSegment());
- return res.isHasBaseTableIndexData();
- }
-
- private void removeIndexesFromSegments(String segmentId, long indexId) {
- removeIndexesFromSegments(ImmutableList.of(segmentId), indexId);
- }
-
- private void removeIndexesFromSegments(List<String> segmentIds, long indexId) {
- modelService.removeIndexesFromSegments(getProject(), modelId, segmentIds, ImmutableList.of(indexId));
- waitAllJobFinish();
- }
-
- private void setQuerySession(String catalog, String jdbcUrl, String driverClassName) {
- System.setProperty("kylin.query.use-tableindex-answer-non-raw-query", "true");
- ss.sessionState().conf().setConfString("spark.sql.catalog." + catalog,
- "org.apache.spark.sql.execution.datasources.jdbc.v2.SecondStorageCatalog");
- ss.sessionState().conf().setConfString("spark.sql.catalog." + catalog + ".url", jdbcUrl);
- ss.sessionState().conf().setConfString("spark.sql.catalog." + catalog + ".driver", driverClassName);
- }
-
- private void assertQueryResult(String sql, long hitLayoutId) throws SQLException {
- OLAPContext.clearThreadLocalContexts();
- QueryContext.current().close();
- QueryContext.current().setRetrySecondStorage(true);
- ExecAndComp.queryModel(getProject(), sql);
- assertTrue(OLAPContext.getNativeRealizations().stream().allMatch(NativeQueryRealization::isSecondStorage));
- assertTrue(OLAPContext.getNativeRealizations().stream().findFirst().isPresent());
- assertEquals(OLAPContext.getNativeRealizations().stream().findFirst().get().getLayoutId().longValue(),
- hitLayoutId);
- }
-
- public void cleanSegments(Set<String> segments, Set<Long> layoutIds) {
- SecondStorageUtil.cleanSegments(getProject(), modelId, segments, layoutIds);
-
- val jobHandler = new SecondStorageIndexCleanJobHandler();
- final JobParam param = SecondStorageJobParamUtil.layoutCleanParam(getProject(), modelId, "ADMIN", layoutIds,
- segments);
- ClickHouseUtils.simulateJobMangerAddJob(param, jobHandler);
- waitAllJobFinish();
- }
-
- private void waitAllJobFinish() {
- NExecutableManager.getInstance(getConfig(), getProject()).getAllExecutables()
- .forEach(exec -> waitJobFinish(getProject(), exec.getId()));
- }
-
- private void deleteSegmentById(String segmentId) {
- modelService.deleteSegmentById(modelId, getProject(), new String[] { segmentId }, true);
- }
-
- private void triggerSegmentLoad(List<String> segments) {
- val request = new StorageRequest();
- request.setProject(getProject());
- request.setModel(modelId);
- request.setSegmentIds(segments);
- EnvelopeResponse<List<String>> jobs = secondStorageEndpoint.getAllSecondStorageJobs();
- assertEquals("000", jobs.getCode());
- assertEquals(0, jobs.getData().size());
- EnvelopeResponse<JobInfoResponse> res = secondStorageEndpoint.loadStorage(request);
- assertEquals("000", res.getCode());
- EnvelopeResponse<List<String>> jobs1 = secondStorageEndpoint.getProjectSecondStorageJobs(getProject());
- assertEquals("000", jobs1.getCode());
- assertEquals(1, jobs1.getData().size());
- for (JobInfoResponse.JobInfo job : res.getData().getJobs()) {
- waitJobFinish(getProject(), job.getJobId());
- }
- }
-
- private void triggerSegmentClean(List<String> segments) {
- val request = new StorageRequest();
- request.setProject(getProject());
- request.setModel(modelId);
- request.setSegmentIds(segments);
- secondStorageEndpoint.cleanStorage(request, segments);
- val manager = NExecutableManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val job = manager.getAllExecutables().stream().filter(ClickHouseSegmentCleanJob.class::isInstance).findFirst();
- Assert.assertTrue(job.isPresent());
- waitJobEnd(getProject(), job.get().getId());
- }
-
- @Test
- public void test2Replica2ShardHA() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse3 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse4 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1, clickhouse2, clickhouse3, clickhouse4 }, 2,
- catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(4, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- triggerClickHouseJob(getDataFlow());
-
- {
- long layoutId = getDataFlow().getIndexPlan().getBaseTableLayoutId();
- List<Integer> rowsList = getHAModelRowCount(getProject(), modelId, layoutId,
- new JdbcDatabaseContainer[] { clickhouse1, clickhouse3 });
- Assert.assertEquals(1, rowsList.stream().distinct().count());
- rowsList = getHAModelRowCount(getProject(), modelId, layoutId,
- new JdbcDatabaseContainer[] { clickhouse2, clickhouse4 });
- Assert.assertEquals(1, rowsList.stream().distinct().count());
- }
-
- String sql = "select CAL_DT from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-02'";
- Map<String, Map<String, Boolean>> nodeStatusMap;
-
- {
- // all live
- clearQueryContext();
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertTrue(OLAPContext.getNativeRealizations().stream()
- .allMatch(NativeQueryRealization::isSecondStorage));
- }
-
- {
- // all group down
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false, "node02", false),
- "pair1", ImmutableMap.of("node01", false, "node03", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertTrue(OLAPContext.getNativeRealizations().stream()
- .noneMatch(NativeQueryRealization::isSecondStorage));
- }
-
- {
- // different group down
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false, "node02", true),
- "pair1", ImmutableMap.of("node01", false, "node03", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- clearQueryContext();
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertTrue(OLAPContext.getNativeRealizations().stream()
- .allMatch(NativeQueryRealization::isSecondStorage));
- }
-
- {
- // same group down
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true, "node02", false),
- "pair1", ImmutableMap.of("node01", false, "node03", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertTrue(OLAPContext.getNativeRealizations().stream()
- .anyMatch(NativeQueryRealization::isSecondStorage));
- }
-
- {
- // down one shard
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false, "node02", false),
- "pair1", ImmutableMap.of("node01", true, "node03", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertTrue(OLAPContext.getNativeRealizations().stream()
- .noneMatch(NativeQueryRealization::isSecondStorage));
-
- }
-
- {
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false, "node02", true),
- "pair1", ImmutableMap.of("node01", true, "node03", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- assertEquals("000", secondStorageEndpoint.tableSync(getProject()).getCode());
- getTableFlow().getTableDataList()
- .forEach(tableData -> tableData.getPartitions().forEach(partition -> {
- assertEquals(2, partition.getShardNodes().size());
- if (partition.getShardNodes().contains("node00")) {
- assertEquals(0L, partition.getSizeInNode().get("node00").longValue());
- }
- }));
- }
-
- {
- val lock1 = new ProjectLockOperateRequest();
- lock1.setProject(getProject());
- lock1.setLockTypes(Arrays.asList(LockTypeEnum.LOAD.name(), LockTypeEnum.QUERY.name()));
- lock1.setOperateType(LockOperateTypeEnum.LOCK.name());
- ClickHouseSimpleITTestUtils.checkLockOperateResult(secondStorageEndpoint.lockOperate(lock1),
- Arrays.asList(LockTypeEnum.LOAD.name(), LockTypeEnum.QUERY.name()), getProject());
-
- val lock2 = new ProjectLockOperateRequest();
- lock2.setProject(getProject());
- lock2.setLockTypes(Collections.singletonList(LockTypeEnum.LOAD.name()));
- lock2.setOperateType(LockOperateTypeEnum.UNLOCK.name());
- ClickHouseSimpleITTestUtils.checkLockOperateResult(secondStorageEndpoint.lockOperate(lock2),
- Collections.singletonList(LockTypeEnum.QUERY.name()), getProject());
-
- secondStorageService.sizeInNode(getProject());
-
- val lock3 = new ProjectLockOperateRequest();
- lock3.setProject(getProject());
- lock3.setLockTypes(Collections.singletonList(LockTypeEnum.QUERY.name()));
- lock3.setOperateType(LockOperateTypeEnum.UNLOCK.name());
- ClickHouseSimpleITTestUtils.checkLockOperateResult(secondStorageEndpoint.lockOperate(lock3),
- Collections.emptyList(), getProject());
- }
-
- {
- // test recover model
- val request = new RecoverRequest();
- request.setProject(getProject());
- val response = openSecondStorageEndpoint.recoverProject(request);
- Assert.assertEquals(1, response.getData().getSubmittedModels().size());
- Assert.assertEquals(0, response.getData().getFailedModels().size());
- }
-
- //reset status
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true, "node02", true),
- "pair1", ImmutableMap.of("node01", true, "node03", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- return true;
- });
- }
- }
-
- @Test
- public void test1Replica2ShardHA() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- val container = new JdbcDatabaseContainer[] { clickhouse1, clickhouse2 };
- int replica = 1;
- configClickhouseWith(container, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(2, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- triggerClickHouseJob(getDataFlow());
-
- String sql = "select CAL_DT from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-02'";
- Map<String, Map<String, Boolean>> nodeStatusMap;
-
- testForceToTSAndChDown(sql, container, replica);
-
- {
- // testGroupNodeDownForceToTierStorageOK
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false), "pair1",
- ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- ExecAndComp.queryModel(getProject(), sql, null);
- Assert.assertFalse(OLAPContext.getNativeRealizations().stream()
- .allMatch(NativeQueryRealization::isSecondStorage));
- }
-
- {
- // testForceToTierStoragePushDown
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false), "pair1",
- ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_PUSH_DOWN);
- queryContext.setForceTableIndex(false);
- assertThrows(SQLException.class, () -> ExecAndComp.queryModel(getProject(), sql));
- }
-
- {
- // testForceToTierStorageInvalidParameters
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false), "pair1",
- ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_PUSH_DOWN);
- queryContext.setForceTableIndex(true);
- assertThrows(SQLException.class, () -> ExecAndComp.queryModel(getProject(), sql));
- }
-
- {
- // testeDownForceToTierStorageReturnError
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false), "pair1",
- ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_RETURN);
- assertThrows(SQLException.class, () -> ExecAndComp.queryModel(getProject(), sql));
- }
-
- {
- // testForceToTierStorageOtherValue
- clearQueryContext();
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false), "pair1",
- ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TAIL);
- assertThrows(SQLException.class, () -> ExecAndComp.queryModel(getProject(), sql));
- }
-
- {
- // testReverseForceToTierStorageWhenCHDataNotComplete
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true), "pair1",
- ImmutableMap.of("node01", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- triggerSegmentClean(
- getDataFlow().getSegments().stream().map(NDataSegment::getId).collect(Collectors.toList()));
- val tablePartitions = getTableFlow().getTableDataList().get(0).getPartitions();
- assertTrue(tablePartitions.isEmpty());
-
- for (ForceToTieredStorage f : ForceToTieredStorage.values()) {
- clearQueryContext();
- QueryContext.current().setForcedToTieredStorage(f);
-
- try {
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertFalse(OLAPContext.getNativeRealizations().stream()
- .allMatch(NativeQueryRealization::isSecondStorage));
- } catch (Exception e) {
- assertTrue(e instanceof SQLException);
- }
- }
- triggerClickHouseJob(getDataFlow());
- }
-
- testReverseForceToTierStorageWhenCHUnavailable(sql);
- testReverseForceToTierStorageWhenCHOK(sql);
-
- //reset status
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true), "pair1",
- ImmutableMap.of("node01", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
-
- testForceToTierStorageShutTierStorage(sql);
- return true;
- });
- }
- }
-
- @SneakyThrows
- private void testForceToTSAndChDown(String sql, JdbcDatabaseContainer<?>[] container, int replica) {
- ExecAndComp.queryModel(getProject(), sql);
- OLAPContext.getNativeRealizations().stream().findFirst().ifPresent(r -> assertTrue(r.isSecondStorage()));
-
- for (JdbcDatabaseContainer<?> clickhouse : container) {
- clickhouse.stop();
- }
- clearQueryContext();
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_RETURN);
- queryContext.setForceTableIndex(true);
- assertThrows(SQLException.class, () -> ExecAndComp.queryModel(getProject(), sql));
-
- clearQueryContext();
- queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_DFS);
- queryContext.setForceTableIndex(false);
- ExecAndComp.queryModel(getProject(), sql);
-
- for (JdbcDatabaseContainer<?> clickhouse : container) {
- clickhouse.start();
- }
- ClickHouseUtils.internalConfigClickHouse(container, replica);
- }
-
- private void testReverseForceToTierStorageWhenCHUnavailable(String sql) {
- // testReverseForceToTierStorageWhenCHUnavailable
- clearQueryContext();
- Map<String, Map<String, Boolean>> nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", false),
- "pair1", ImmutableMap.of("node01", false));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- List<Boolean> reverseForceTableIndex = Arrays.asList(true, false);
- for (ForceToTieredStorage f : ForceToTieredStorage.values()) {
- for (Boolean forceTableIndex : reverseForceTableIndex) {
- clearQueryContext();
- QueryContext.current().setForceTableIndex(forceTableIndex);
- QueryContext.current().setForcedToTieredStorage(f);
- try {
- ExecAndComp.queryModel(getProject(), sql);
- Assert.assertFalse(OLAPContext.getNativeRealizations().stream()
- .allMatch(NativeQueryRealization::isSecondStorage));
- } catch (Exception e) {
- assertTrue(e instanceof SQLException);
- }
- }
- }
- }
-
- private void testReverseForceToTierStorageWhenCHOK(String sql) throws SQLException {
- // testReverseForceToTierStorageWhenCHOK
- Map<String, Map<String, Boolean>> nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true),
- "pair1", ImmutableMap.of("node01", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- for (ForceToTieredStorage f : ForceToTieredStorage.values()) {
- clearQueryContext();
- QueryContext.current().setForcedToTieredStorage(f);
- ExecAndComp.queryModel(getProject(), sql);
- }
- }
-
- private void testForceToTierStorageShutTierStorage(String sql) throws SQLException {
- // testForceToTierStorageShutTierStorage
- secondStorageService.changeProjectSecondStorageState(getProject(), SecondStorageNodeHelper.getAllPairs(),
- false);
- clearQueryContext();
- QueryContext queryContext = QueryContext.current();
- queryContext.setForcedToTieredStorage(ForceToTieredStorage.CH_FAIL_TO_PUSH_DOWN);
- Dataset<Row> rows = ExecAndComp.queryModel(getProject(), sql, null);
- Assert.assertFalse(
- OLAPContext.getNativeRealizations().stream().allMatch(NativeQueryRealization::isSecondStorage));
- }
-
- @Test
- public void test1Replica3Shard() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse3 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1, clickhouse2 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(2, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- triggerClickHouseJob(getDataFlow());
-
- {
- getTableFlow().getTableDataList().forEach(tableData -> tableData.getPartitions()
- .forEach(partition -> assertEquals(2, partition.getShardNodes().size())));
- }
-
- ClickHouseUtils.internalConfigClickHouse(
- new JdbcDatabaseContainer[] { clickhouse1, clickhouse2, clickhouse3 }, 1);
- secondStorageService.changeProjectSecondStorageState(getProject(), ImmutableList.of("pair2"), true);
- Assert.assertEquals(3, SecondStorageUtil.listProjectNodes(getProject()).size());
-
- Map<String, Map<String, Boolean>> nodeStatusMap;
-
- {
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true), "pair1",
- ImmutableMap.of("node01", false), "pair2", ImmutableMap.of("node02", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- EnvelopeResponse<ProjectTableSyncResponse> response = secondStorageEndpoint.tableSync(getProject());
- assertEquals("000", response.getCode());
- getTableFlow().getTableDataList()
- .forEach(tableData -> tableData.getPartitions().forEach(partition -> {
- assertEquals(3, partition.getShardNodes().size());
- assertEquals(0, partition.getSizeInNode().get("node01").longValue());
- assertEquals(0, partition.getSizeInNode().get("node02").longValue());
- }));
- }
-
- //reset status
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true), "pair1",
- ImmutableMap.of("node01", true), "pair2", ImmutableMap.of("node02", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- return true;
- });
- }
- }
-
- @Test
- public void test1Replica1Shard() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1 }, 1, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(1, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
- String sql = "select CAL_DT from TEST_KYLIN_FACT where CAL_DT between '2012-01-01' and '2012-01-08'";
- List<Long> resultList = new ArrayList<>();
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- clearQueryContext();
- resultList.add(ExecAndComp.queryModel(getProject(), sql).count());
- triggerClickHouseJob(getDataFlow());
- clearQueryContext();
- resultList.add(ExecAndComp.queryModel(getProject(), sql).count());
-
- wrapWithKylinConfigMode("ANY", "CHAIN", () -> {
- modelBuildService.refreshSegmentById(
- new RefreshSegmentParams(getProject(), modelId, getDataFlow().getSegments().stream()
- .map(NDataSegment::getId).collect(Collectors.toList()).toArray(new String[] {})));
- waitAllJobFinish();
- clearQueryContext();
- resultList.add(ExecAndComp.queryModel(getProject(), sql).count());
- triggerClickHouseJob(getDataFlow());
- clearQueryContext();
- resultList.add(ExecAndComp.queryModel(getProject(), sql).count());
- return true;
- });
-
- assertEquals(1, resultList.stream().distinct().count());
- return true;
- });
- }
- }
-
- @Test
- public void test2Replica1Shard() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse();
- JdbcDatabaseContainer<?> clickhouse2 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
- configClickhouseWith(new JdbcDatabaseContainer[] { clickhouse1, clickhouse2 }, 2, catalog, () -> {
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(2, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse1.getJdbcUrl(), clickhouse1.getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-01", "2012-01-02");
- waitAllJobFinish();
- testWait();
- triggerClickHouseJob(getDataFlow());
- Map<String, Map<String, Boolean>> nodeStatusMap;
-
- {
- testSecondStorageQueryRandom();
- }
-
- {
- Date now = new Date();
- List<NDataSegmentResponse> mockSegments = Lists.newArrayList();
- NDataSegmentResponse segmentResponse1 = new NDataSegmentResponse();
- segmentResponse1.setId("1");
- segmentResponse1.setRowCount(1);
- segmentResponse1.setCreateTime(DateUtils.addHours(now, -1).getTime());
-
- NDataSegmentResponse segmentResponse2 = new NDataSegmentResponse();
- segmentResponse2.setId("2");
- segmentResponse2.setRowCount(2);
- segmentResponse2.setCreateTime(now.getTime());
-
- NDataSegmentResponse segmentResponse3 = new NDataSegmentResponse();
- segmentResponse3.setId("3");
- segmentResponse3.setRowCount(3);
- segmentResponse3.setCreateTime(DateUtils.addHours(now, 1).getTime());
-
- String id = getTableFlow().getTableDataList().get(0).getPartitions().get(0).getSegmentId();
- segmentResponse1.setId(id);
- segmentResponse2.setId(id);
- segmentResponse3.setId(id);
-
- mockSegments.add(segmentResponse1);
- mockSegments.add(segmentResponse3);
- mockSegments.add(segmentResponse2);
-
- modelService.addSecondStorageResponse(modelId, getProject(), mockSegments, getDataFlow());
- Assert.assertEquals(2, (mockSegments.get(0)).getSecondStorageNodes().get("pair0").size());
- }
-
- {
- try (ClickHouse clickHouse = new ClickHouse(SecondStorageNodeHelper.resolve("node01"))) {
- ShowDatabases showDatabases = new ShowDatabases();
- List<String> databases = clickHouse.query(showDatabases.toSql(),
- ShowDatabasesParser.SHOW_DATABASES);
- databases = databases.stream()
- .filter(database -> !database.equals("default") && !database.equals("system"))
- .collect(Collectors.toList());
- assertTrue(new HashSet<>(databases).contains("UT_table_index_incremental"));
- List<String> tables = clickHouse.query(
- ShowTables.createShowTables("UT_table_index_incremental").toSql(),
- ShowDatabasesParser.SHOW_DATABASES);
- assertTrue(new HashSet<>(tables).contains("acfde546_2cc9_4eec_bc92_e3bd46d4e2ee_20000000001"));
- }
-
- assertEquals("000", secondStorageEndpoint.tableSync(getProject()).getCode());
- SecondStorageMetadataRequest request = new SecondStorageMetadataRequest();
- request.setProject(getProject());
- assertEquals("000", secondStorageEndpoint.sizeInNode(request).getCode());
- }
-
- //reset status
- nodeStatusMap = ImmutableMap.of("pair0", ImmutableMap.of("node00", true, "node02", true));
- secondStorageEndpoint.updateNodeStatus(nodeStatusMap);
- return true;
- });
- }
- }
-
- @Test
- public void testClickhouseLowCardinality() throws Exception {
- try (JdbcDatabaseContainer<?> clickhouse1 = ClickHouseUtils.startClickHouse()) {
- final String catalog = "default";
-
- Unsafe.setProperty(ClickHouseLoad.SOURCE_URL, getSourceUrl());
- Unsafe.setProperty(ClickHouseLoad.ROOT_PATH, getLocalWorkingDirectory());
-
- val clickhouse = new JdbcDatabaseContainer[] { clickhouse1 };
- int replica = 1;
- configClickhouseWith(clickhouse, replica, catalog, () -> {
- QueryOperator queryOperator = SecondStorageFactoryUtils.createQueryMetricOperator(getProject());
- queryOperator.modifyColumnByCardinality("default", "table", Sets.newHashSet());
-
- buildIncrementalLoadQuery("2012-01-02", "2012-01-03");
- waitAllJobFinish();
- secondStorageService.changeProjectSecondStorageState(getProject(),
- SecondStorageNodeHelper.getAllPairs(), true);
- Assert.assertEquals(clickhouse.length, SecondStorageUtil.listProjectNodes(getProject()).size());
- secondStorageService.changeModelSecondStorageState(getProject(), modelId, true);
- setQuerySession(catalog, clickhouse[0].getJdbcUrl(), clickhouse[0].getDriverClassName());
-
- buildIncrementalLoadQuery("2012-01-04", "2012-01-05");
- waitAllJobFinish();
- secondStorageScheduleService.secondStorageLowCardinality();
- EpochManager epochManager = EpochManager.getInstance();
- String ownerIdentity = EpochOrchestrator.getOwnerIdentity();
- Epoch epoch = epochManager.getEpoch(EpochManager.GLOBAL);
- if (epoch == null) {
- epoch = new Epoch(1L, EpochManager.GLOBAL, ownerIdentity, System.currentTimeMillis(), "all", null,
- 0L);
- }
- ReflectionTestUtils.invokeMethod(epochManager, "insertOrUpdateEpoch", epoch);
- secondStorageScheduleService.secondStorageLowCardinality();
-
- val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), getProject());
- val dataflow = dataflowManager.getDataflow(modelId);
-
- String rows = "";
- String database = NameUtil.getDatabase(dataflow);
- val destTableName = NameUtil.getTable(dataflow, SecondStorageUtil.getBaseIndex(dataflow).getId());
-
- 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);
-
- getConfig().setProperty("kylin.second-storage.high-cardinality-number", "2");
- secondStorageScheduleService.secondStorageLowCardinality();
- 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(NULLABLE_STRING, rows);
- try {
- secondStorageService.modifyColumn(getProject(), modelId, "TEST_KYLIN_FACT.LSTG_FORMAT",
- LOW_CARDINALITY_STRING);
- } catch (KylinException e) {
- Assert.assertEquals(INVALID_PARAMETER.toErrorCode(), e.getErrorCode());
- }
- secondStorageService.modifyColumn(getProject(), modelId, "TEST_KYLIN_FACT.LSTG_FORMAT_NAME",
- LOW_CARDINALITY_STRING);
- 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);
-
- 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;
- });
- }
- }
-
- public List<Integer> getHAModelRowCount(String project, String modelId, long layoutId,
- JdbcDatabaseContainer<?>[] clickhouse) {
- KylinConfig config = KylinConfig.getInstanceFromEnv();
- val database = NameUtil.getDatabase(config, project);
- val table = NameUtil.getTable(modelId, layoutId);
-
- return Arrays.stream(clickhouse).map(JdbcDatabaseContainer::getJdbcUrl).map(url -> {
- try (ClickHouse clickHouse = new ClickHouse(url)) {
- List<Integer> count = clickHouse.query("select count(*) from `" + database + "`.`" + table + "`",
- rs -> {
- try {
- return rs.getInt(1);
- } catch (SQLException e) {
- assertNull(e);
- return 0;
- }
- });
- Assert.assertFalse(count.isEmpty());
- return count.get(0);
- } catch (Exception e) {
- assertNull(e);
- return 0;
- }
- }).collect(Collectors.toList());
- }
-
- private void testWait() {
- val executableManager = NExecutableManager.getInstance(getConfig(), getProject());
- val job = executableManager.getAllExecutables().stream().filter(j -> j instanceof NSparkCubingJob).findFirst();
- Map<String, String> waiteTimeMap = new HashMap<>();
- job.ifPresent(j -> ((NSparkCubingJob) j).getTasks().forEach(task -> {
- final ExecutableStepResponse executableStepResponse = jobService.parseToExecutableStep(task,
- executableManager.getOutput(task.getId()), waiteTimeMap, j.getOutput().getState());
- assertTrue(executableStepResponse.getWaitTime() >= 0L);
- }));
- }
-
- private void clearQueryContext() {
- OLAPContext.clearThreadLocalContexts();
- QueryContext.current().close();
- QueryContext.current().setRetrySecondStorage(true);
- }
-}
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 ff8ed15e2d..210adb7739 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
@@ -461,7 +461,7 @@ public class SecondStorageService extends BasicService implements SecondStorageU
val dataflowManager = NDataflowManager.getInstance(config, project);
val segIds = dataflowManager.getDataflow(model).getQueryableSegments().stream()
.map(NDataSegment::getId).collect(Collectors.toList());
- return modelService.exportSegmentToSecondStorage(project, model, segIds.toArray(new String[]{}));
+ return null;
}
@Transaction(project = 0)
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCleanupAfterMergeStep.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCleanupAfterMergeStep.java
index adcc16c0a4..90803f41f0 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCleanupAfterMergeStep.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCleanupAfterMergeStep.java
@@ -18,7 +18,7 @@
package org.apache.kylin.engine.spark.job;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.KylinConfig;
@@ -59,7 +59,7 @@ public class NSparkCleanupAfterMergeStep extends NSparkExecutable {
for (String segmentId : segmentIds) {
String path = dataflow.getSegmentHdfsPath(segmentId);
- if (!SecondStorageUtil.isModelEnable(dataflow.getProject(), dataflow.getModel().getUuid())) {
+// if (!SecondStorageUtil.isModelEnable(dataflow.getProject(), dataflow.getModel().getUuid())) {
if (!timeMachineEnabled) {
try {
HadoopUtil.deletePath(HadoopUtil.getCurrentConfiguration(), new Path(path));
@@ -70,10 +70,10 @@ public class NSparkCleanupAfterMergeStep extends NSparkExecutable {
+ " Please try workaround thru garbage clean manually.", segmentId, name, e);
}
}
- } else {
- logger.info("ClickHouse is enabled for the model, please delete segments {} in dataflow {} manually.", //
- segmentIds, name);
- }
+// } else {
+// logger.info("ClickHouse is enabled for the model, please delete segments {} in dataflow {} manually.", //
+// segmentIds, name);
+// }
}
return ExecuteResult.createSucceed();
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCubingJob.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCubingJob.java
index 18d52bf851..0b98bbc6b9 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCubingJob.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkCubingJob.java
@@ -18,14 +18,14 @@
package org.apache.kylin.engine.spark.job;
-import io.kyligence.kap.secondstorage.SecondStorageConstants;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
-import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
+//import io.kyligence.kap.secondstorage.SecondStorageConstants;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.val;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
+//import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.KylinConfigExt;
import org.apache.kylin.common.exception.JobErrorCode;
@@ -38,7 +38,7 @@ import org.apache.kylin.guava30.shaded.common.collect.Sets;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.DefaultExecutableOnModel;
import org.apache.kylin.job.execution.ExecutableParams;
-import org.apache.kylin.job.execution.JobSchedulerModeEnum;
+//import org.apache.kylin.job.execution.JobSchedulerModeEnum;
import org.apache.kylin.job.execution.JobTypeEnum;
import org.apache.kylin.job.factory.JobFactory;
import org.apache.kylin.metadata.cube.model.IndexPlan;
@@ -55,11 +55,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collections;
+//import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Map;
-import java.util.Objects;
+//import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
@@ -191,58 +191,58 @@ public class NSparkCubingJob extends DefaultExecutableOnModel {
AbstractExecutable resourceDetect = JobStepType.RESOURCE_DETECT.createStep(job, config);
AbstractExecutable cubing = JobStepType.CUBING.createStep(job, config);
AbstractExecutable updateMetadata = JobStepType.UPDATE_METADATA.createStep(job, config);
- AbstractExecutable secondStorageDeleteIndex = initSecondStorageDeleteIndex(params.getToBeDeletedLayouts(),
- jobType, df, job, config);
- AbstractExecutable secondStorage = initSecondStorage(layouts, jobType, df, job, config);
+// AbstractExecutable secondStorageDeleteIndex = initSecondStorageDeleteIndex(params.getToBeDeletedLayouts(),
+// jobType, df, job, config);
+// AbstractExecutable secondStorage = initSecondStorage(layouts, jobType, df, job, config);
AbstractExecutable cleanUpTransactionalTable = initCleanUpTransactionalTable(kylinConfig, df, job, config);
- if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
- setDAGRelations(job, config, new NSparkCubingJob.NSparkCubingJobStep(resourceDetect, cubing, updateMetadata,
- secondStorageDeleteIndex, secondStorage, cleanUpTransactionalTable));
- }
+// if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
+// setDAGRelations(job, config, new NSparkCubingJob.NSparkCubingJobStep(resourceDetect, cubing, updateMetadata,
+// secondStorageDeleteIndex, secondStorage, cleanUpTransactionalTable));
+// }
return job;
}
- private static AbstractExecutable initSecondStorageDeleteIndex(Set<LayoutEntity> toBeDeletedLayouts,
- JobTypeEnum jobType, NDataflow df, NSparkCubingJob job, KylinConfigExt config) {
- if (!SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
- return null;
- }
-
- AbstractExecutable secondStorage = null;
- if (Objects.equals(jobType, JobTypeEnum.INDEX_BUILD) && CollectionUtils.isNotEmpty(toBeDeletedLayouts)) {
- secondStorage = JobStepType.SECOND_STORAGE_INDEX_CLEAN.createStep(job, config);
- }
- return secondStorage;
- }
-
- private static AbstractExecutable initSecondStorage(Set<LayoutEntity> layouts, JobTypeEnum jobType, NDataflow df,
- NSparkCubingJob job, KylinConfigExt config) {
- AbstractExecutable secondStorage = null;
- if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
- // can't refresh segment when second storage do rebalanced
- if (Objects.equals(jobType, JobTypeEnum.INDEX_REFRESH)) {
- SecondStorageUtil.validateProjectLock(df.getProject(),
- Collections.singletonList(LockTypeEnum.LOAD.name()));
- }
- boolean hasBaseIndex = layouts.stream().anyMatch(SecondStorageUtil::isBaseTableIndex);
- if (Objects.equals(jobType, JobTypeEnum.INDEX_BUILD) || Objects.equals(jobType, JobTypeEnum.INC_BUILD)) {
- if (hasBaseIndex) {
- secondStorage = JobStepType.SECOND_STORAGE_EXPORT.createStep(job, config);
- }
- } else if (Objects.equals(jobType, JobTypeEnum.INDEX_REFRESH) && hasBaseIndex) {
- val oldSegs = job.getTargetSegments().stream().map(segId -> {
- val curSeg = df.getSegment(segId);
- return Objects.requireNonNull(df.getSegments().stream()
- .filter(seg -> seg.getSegRange().equals(curSeg.getSegRange()) && !seg.getId().equals(segId))
- .findFirst().orElse(null)).getId();
- }).collect(Collectors.toList());
- job.setParam(SecondStorageConstants.P_OLD_SEGMENT_IDS, String.join(",", oldSegs));
- secondStorage = JobStepType.SECOND_STORAGE_REFRESH.createStep(job, config);
- }
- }
- return secondStorage;
- }
+// private static AbstractExecutable initSecondStorageDeleteIndex(Set<LayoutEntity> toBeDeletedLayouts,
+// JobTypeEnum jobType, NDataflow df, NSparkCubingJob job, KylinConfigExt config) {
+// if (!SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
+// return null;
+// }
+//
+// AbstractExecutable secondStorage = null;
+// if (Objects.equals(jobType, JobTypeEnum.INDEX_BUILD) && CollectionUtils.isNotEmpty(toBeDeletedLayouts)) {
+// secondStorage = JobStepType.SECOND_STORAGE_INDEX_CLEAN.createStep(job, config);
+// }
+// return secondStorage;
+// }
+
+// private static AbstractExecutable initSecondStorage(Set<LayoutEntity> layouts, JobTypeEnum jobType, NDataflow df,
+// NSparkCubingJob job, KylinConfigExt config) {
+// AbstractExecutable secondStorage = null;
+// if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
+// // can't refresh segment when second storage do rebalanced
+// if (Objects.equals(jobType, JobTypeEnum.INDEX_REFRESH)) {
+// SecondStorageUtil.validateProjectLock(df.getProject(),
+// Collections.singletonList(LockTypeEnum.LOAD.name()));
+// }
+// boolean hasBaseIndex = layouts.stream().anyMatch(SecondStorageUtil::isBaseTableIndex);
+// if (Objects.equals(jobType, JobTypeEnum.INDEX_BUILD) || Objects.equals(jobType, JobTypeEnum.INC_BUILD)) {
+// if (hasBaseIndex) {
+// secondStorage = JobStepType.SECOND_STORAGE_EXPORT.createStep(job, config);
+// }
+// } else if (Objects.equals(jobType, JobTypeEnum.INDEX_REFRESH) && hasBaseIndex) {
+// val oldSegs = job.getTargetSegments().stream().map(segId -> {
+// val curSeg = df.getSegment(segId);
+// return Objects.requireNonNull(df.getSegments().stream()
+// .filter(seg -> seg.getSegRange().equals(curSeg.getSegRange()) && !seg.getId().equals(segId))
+// .findFirst().orElse(null)).getId();
+// }).collect(Collectors.toList());
+// job.setParam(SecondStorageConstants.P_OLD_SEGMENT_IDS, String.join(",", oldSegs));
+// secondStorage = JobStepType.SECOND_STORAGE_REFRESH.createStep(job, config);
+// }
+// }
+// return secondStorage;
+// }
private static AbstractExecutable initCleanUpTransactionalTable(KylinConfig kylinConfig, NDataflow df,
NSparkCubingJob job, KylinConfigExt config) {
@@ -259,31 +259,31 @@ public class NSparkCubingJob extends DefaultExecutableOnModel {
return cleanUpTransactionalTable;
}
- public static void setDAGRelations(AbstractExecutable job, KylinConfig config,
- NSparkCubingJob.NSparkCubingJobStep jobStep) {
- if (!StringUtils.equalsIgnoreCase(config.getJobSchedulerMode(), JobSchedulerModeEnum.CHAIN.toString())) {
- AbstractExecutable resourceDetect = jobStep.getResourceDetect();
- AbstractExecutable cubing = jobStep.getCubing();
- AbstractExecutable updateMetadata = jobStep.getUpdateMetadata();
- AbstractExecutable secondStorageDeleteIndex = jobStep.getSecondStorageDeleteIndex();
- AbstractExecutable secondStorage = jobStep.getSecondStorage();
- AbstractExecutable cleanUpTransactionalTable = jobStep.getCleanUpTransactionalTable();
-
- initResourceDetectDagNode(resourceDetect, cubing, secondStorage);
- cubing.setNextSteps(Sets.newHashSet(updateMetadata.getId()));
- updateMetadata.setPreviousStep(cubing.getId());
- AbstractExecutable preStep = updateMetadata;
- if (secondStorageDeleteIndex != null) {
- setNextStep(preStep, secondStorageDeleteIndex);
- preStep = secondStorageDeleteIndex;
- }
- if (cleanUpTransactionalTable != null) {
- preStep.setNextSteps(Sets.newHashSet(cleanUpTransactionalTable.getId()));
- cleanUpTransactionalTable.setParentId(preStep.getId());
- }
- job.setJobSchedulerMode(JobSchedulerModeEnum.DAG);
- }
- }
+// public static void setDAGRelations(AbstractExecutable job, KylinConfig config,
+// NSparkCubingJob.NSparkCubingJobStep jobStep) {
+// if (!StringUtils.equalsIgnoreCase(config.getJobSchedulerMode(), JobSchedulerModeEnum.CHAIN.toString())) {
+// AbstractExecutable resourceDetect = jobStep.getResourceDetect();
+// AbstractExecutable cubing = jobStep.getCubing();
+// AbstractExecutable updateMetadata = jobStep.getUpdateMetadata();
+// AbstractExecutable secondStorageDeleteIndex = jobStep.getSecondStorageDeleteIndex();
+// AbstractExecutable secondStorage = jobStep.getSecondStorage();
+// AbstractExecutable cleanUpTransactionalTable = jobStep.getCleanUpTransactionalTable();
+//
+// initResourceDetectDagNode(resourceDetect, cubing, secondStorage);
+// cubing.setNextSteps(Sets.newHashSet(updateMetadata.getId()));
+// updateMetadata.setPreviousStep(cubing.getId());
+// AbstractExecutable preStep = updateMetadata;
+// if (secondStorageDeleteIndex != null) {
+// setNextStep(preStep, secondStorageDeleteIndex);
+// preStep = secondStorageDeleteIndex;
+// }
+// if (cleanUpTransactionalTable != null) {
+// preStep.setNextSteps(Sets.newHashSet(cleanUpTransactionalTable.getId()));
+// cleanUpTransactionalTable.setParentId(preStep.getId());
+// }
+// job.setJobSchedulerMode(JobSchedulerModeEnum.DAG);
+// }
+// }
private static void setNextStep(AbstractExecutable preStep, AbstractExecutable currentStep) {
preStep.setNextSteps(Sets.newHashSet(currentStep.getId()));
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkMergingJob.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkMergingJob.java
index 977ed0f228..d3599aa1be 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkMergingJob.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/NSparkMergingJob.java
@@ -21,16 +21,16 @@ package org.apache.kylin.engine.spark.job;
import org.apache.kylin.guava30.shaded.common.base.Preconditions;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
-import io.kyligence.kap.secondstorage.SecondStorageConstants;
-import io.kyligence.kap.secondstorage.SecondStorageUtil;
-import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
+//import io.kyligence.kap.secondstorage.SecondStorageConstants;
+//import io.kyligence.kap.secondstorage.SecondStorageUtil;
+//import io.kyligence.kap.secondstorage.enums.LockTypeEnum;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
+//import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.DefaultExecutableOnModel;
import org.apache.kylin.job.execution.ExecutableParams;
-import org.apache.kylin.job.execution.JobSchedulerModeEnum;
+//import org.apache.kylin.job.execution.JobSchedulerModeEnum;
import org.apache.kylin.job.execution.JobTypeEnum;
import org.apache.kylin.job.factory.JobFactory;
import org.apache.kylin.metadata.cube.model.LayoutEntity;
@@ -46,7 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collections;
+//import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -140,42 +140,42 @@ public class NSparkMergingJob extends DefaultExecutableOnModel {
final Segments<NDataSegment> mergingSegments = df.getMergingSegments(mergedSegment);
cleanStep.setParam(NBatchConstants.P_SEGMENT_IDS,
String.join(",", NSparkCubingUtil.toSegmentIds(mergingSegments)));
- AbstractExecutable mergeStep = initSecondMergeStep(mergedSegment, layouts, df, job, config, mergingSegments);
+// AbstractExecutable mergeStep = initSecondMergeStep(mergedSegment, layouts, df, job, config, mergingSegments);
AbstractExecutable updateMetadata = JobStepType.UPDATE_METADATA.createStep(job, config);
- if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
- setDAGRelations(job, config, resourceDetect, merging, cleanStep, mergeStep, updateMetadata);
- }
+// if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())) {
+// setDAGRelations(job, config, resourceDetect, merging, cleanStep, mergeStep, updateMetadata);
+// }
return job;
}
- private static AbstractExecutable initSecondMergeStep(NDataSegment mergedSegment, Set<LayoutEntity> layouts,
- NDataflow df, NSparkMergingJob job, KylinConfig config, Segments<NDataSegment> mergingSegments) {
- AbstractExecutable mergeStep = null;
- if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())
- && layouts.stream().anyMatch(SecondStorageUtil::isBaseTableIndex)) {
- // can't merge segment when second storage do rebalanced
- SecondStorageUtil.validateProjectLock(df.getProject(), Collections.singletonList(LockTypeEnum.LOAD.name()));
- mergeStep = JobStepType.SECOND_STORAGE_MERGE.createStep(job, config);
- mergeStep.setParam(SecondStorageConstants.P_MERGED_SEGMENT_ID, mergedSegment.getId());
- mergeStep.setParam(NBatchConstants.P_SEGMENT_IDS,
- String.join(",", NSparkCubingUtil.toSegmentIds(mergingSegments)));
- }
- return mergeStep;
- }
-
- public static void setDAGRelations(AbstractExecutable job, KylinConfig config, AbstractExecutable resourceDetect,
- AbstractExecutable merging, AbstractExecutable clean, AbstractExecutable secondStorageMerge,
- AbstractExecutable updateMetadata) {
- if (!StringUtils.equalsIgnoreCase(config.getJobSchedulerMode(), JobSchedulerModeEnum.CHAIN.toString())) {
- initResourceDetectDagNode(resourceDetect, merging, secondStorageMerge);
- merging.setNextSteps(Sets.newHashSet(clean.getId()));
- clean.setPreviousStep(merging.getId());
- clean.setNextSteps(Sets.newHashSet(updateMetadata.getId()));
- updateMetadata.setPreviousStep(clean.getId());
- job.setJobSchedulerMode(JobSchedulerModeEnum.DAG);
- }
- }
+// private static AbstractExecutable initSecondMergeStep(NDataSegment mergedSegment, Set<LayoutEntity> layouts,
+// NDataflow df, NSparkMergingJob job, KylinConfig config, Segments<NDataSegment> mergingSegments) {
+// AbstractExecutable mergeStep = null;
+// if (SecondStorageUtil.isModelEnable(df.getProject(), job.getTargetSubject())
+// && layouts.stream().anyMatch(SecondStorageUtil::isBaseTableIndex)) {
+// // can't merge segment when second storage do rebalanced
+// SecondStorageUtil.validateProjectLock(df.getProject(), Collections.singletonList(LockTypeEnum.LOAD.name()));
+// mergeStep = JobStepType.SECOND_STORAGE_MERGE.createStep(job, config);
+// mergeStep.setParam(SecondStorageConstants.P_MERGED_SEGMENT_ID, mergedSegment.getId());
+// mergeStep.setParam(NBatchConstants.P_SEGMENT_IDS,
+// String.join(",", NSparkCubingUtil.toSegmentIds(mergingSegments)));
+// }
+// return mergeStep;
+// }
+
+// public static void setDAGRelations(AbstractExecutable job, KylinConfig config, AbstractExecutable resourceDetect,
+// AbstractExecutable merging, AbstractExecutable clean, AbstractExecutable secondStorageMerge,
+// AbstractExecutable updateMetadata) {
+// if (!StringUtils.equalsIgnoreCase(config.getJobSchedulerMode(), JobSchedulerModeEnum.CHAIN.toString())) {
+// initResourceDetectDagNode(resourceDetect, merging, secondStorageMerge);
+// merging.setNextSteps(Sets.newHashSet(clean.getId()));
+// clean.setPreviousStep(merging.getId());
+// clean.setNextSteps(Sets.newHashSet(updateMetadata.getId()));
+// updateMetadata.setPreviousStep(clean.getId());
+// job.setJobSchedulerMode(JobSchedulerModeEnum.DAG);
+// }
+// }
@Override
public Set<String> getMetadataDumpList(KylinConfig config) {
diff --git a/src/spark-project/sparder/src/main/scala/org/apache/kylin/query/runtime/plan/ResultPlan.scala b/src/spark-project/sparder/src/main/scala/org/apache/kylin/query/runtime/plan/ResultPlan.scala
index a1ad366e6e..b91ae16457 100644
--- a/src/spark-project/sparder/src/main/scala/org/apache/kylin/query/runtime/plan/ResultPlan.scala
+++ b/src/spark-project/sparder/src/main/scala/org/apache/kylin/query/runtime/plan/ResultPlan.scala
@@ -23,7 +23,7 @@ import java.nio.charset.StandardCharsets
import java.util.concurrent.atomic.AtomicLong
import java.{lang, util}
-import io.kyligence.kap.secondstorage.SecondStorageUtil
+//import io.kyligence.kap.secondstorage.SecondStorageUtil
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
import org.apache.commons.io.IOUtils
import org.apache.hadoop.fs.Path
@@ -134,12 +134,12 @@ object ResultPlan extends LogEx {
QueryContext.current().getMetrics.setQueryStageCount(stageCount)
QueryContext.current().getMetrics.setQueryTaskCount(taskCount)
- if (!QueryContext.current().getSecondStorageUsageMap.isEmpty &&
- KylinConfig.getInstanceFromEnv.getSecondStorageQueryMetricCollect) {
- val executedPlan = SecondStorageUtil.collectExecutedPlan(getNormalizedExplain(df))
- val pushedPlan = SecondStorageUtil.convertExecutedPlan(executedPlan, QueryContext.current.getProject, OLAPContext.getNativeRealizations)
- QueryContext.current().getMetrics.setQueryExecutedPlan(pushedPlan)
- }
+// if (!QueryContext.current().getSecondStorageUsageMap.isEmpty &&
+// KylinConfig.getInstanceFromEnv.getSecondStorageQueryMetricCollect) {
+// val executedPlan = SecondStorageUtil.collectExecutedPlan(getNormalizedExplain(df))
+// val pushedPlan = SecondStorageUtil.convertExecutedPlan(executedPlan, QueryContext.current.getProject, OLAPContext.getNativeRealizations)
+// QueryContext.current().getMetrics.setQueryExecutedPlan(pushedPlan)
+// }
logInfo(s"Actual total scan count: $scanRows, " +
s"file scan row count: ${QueryContext.current.getMetrics.getAccumSourceScanRows}, " +
diff --git a/src/spark-project/sparder/src/main/scala/org/apache/spark/sql/KylinDataFrameManager.scala b/src/spark-project/sparder/src/main/scala/org/apache/spark/sql/KylinDataFrameManager.scala
index 115e6c9f28..8de357982d 100644
--- a/src/spark-project/sparder/src/main/scala/org/apache/spark/sql/KylinDataFrameManager.scala
+++ b/src/spark-project/sparder/src/main/scala/org/apache/spark/sql/KylinDataFrameManager.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql
import java.sql.Timestamp
-import io.kyligence.kap.secondstorage.SecondStorage
+//import io.kyligence.kap.secondstorage.SecondStorage
import org.apache.kylin.common.KylinConfig
import org.apache.kylin.metadata.cube.model.{LayoutEntity, NDataflow, NDataflowManager}
import org.apache.kylin.metadata.model.FusionModelManager
@@ -96,10 +96,10 @@ class KylinDataFrameManager(sparkSession: SparkSession) {
}
def read(dataflow: NDataflow, layout: LayoutEntity, pruningInfo: String): DataFrame = {
- SecondStorage.trySecondStorage(sparkSession, dataflow, layout, pruningInfo).getOrElse {
+// SecondStorage.trySecondStorage(sparkSession, dataflow, layout, pruningInfo).getOrElse {
StorageStoreFactory.create(dataflow.getModel.getStorageType)
.read(dataflow, layout, sparkSession, extraOptions.toMap)
- }
+// }
}
/**
diff --git a/src/tool/src/main/java/org/apache/kylin/tool/AbstractInfoExtractorTool.java b/src/tool/src/main/java/org/apache/kylin/tool/AbstractInfoExtractorTool.java
index cbe2286bef..e617eb0744 100644
--- a/src/tool/src/main/java/org/apache/kylin/tool/AbstractInfoExtractorTool.java
+++ b/src/tool/src/main/java/org/apache/kylin/tool/AbstractInfoExtractorTool.java
@@ -604,7 +604,7 @@ public abstract class AbstractInfoExtractorTool extends ExecutableApplication {
protected void exportTieredStorage(String project, File exportDir, long startTime, long endTime, File recordTime) {
Future kgLogTask = executorService.submit(() -> {
recordTaskStartTime(TIERED_STORAGE_LOGS);
- new ClickhouseDiagTool(project).dumpClickHouseServerLog(exportDir, startTime, endTime);
+// new ClickhouseDiagTool(project).dumpClickHouseServerLog(exportDir, startTime, endTime);
recordTaskExecutorTimeToFile(TIERED_STORAGE_LOGS, recordTime);
});
diff --git a/src/tool/src/main/java/org/apache/kylin/tool/ClickhouseDiagTool.java b/src/tool/src/main/java/org/apache/kylin/tool/ClickhouseDiagTool.java
deleted file mode 100644
index d7970e22a1..0000000000
--- a/src/tool/src/main/java/org/apache/kylin/tool/ClickhouseDiagTool.java
+++ /dev/null
@@ -1,240 +0,0 @@
-/*
- * 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.kylin.tool;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Locale;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.zip.GZIPInputStream;
-
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang3.ArrayUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.kylin.common.KylinConfig;
-import org.apache.kylin.common.util.CliCommandExecutor;
-import org.apache.kylin.common.util.Pair;
-import org.joda.time.DateTime;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.kylin.guava30.shaded.common.base.Preconditions;
-
-import io.kyligence.kap.secondstorage.SecondStorage;
-import io.kyligence.kap.secondstorage.SecondStorageNodeHelper;
-import lombok.NoArgsConstructor;
-import lombok.val;
-
-@NoArgsConstructor
-public class ClickhouseDiagTool {
- public static final String SUB_DIR = "tieredStorage";
- private static final Logger logger = LoggerFactory.getLogger("diag");
- //{ck_node_name}_{ip}_{port}
- private static final String CK_NODE_PATH_FORMAT = "%s_%s_%s";
- private static final String SECOND_DATE_FORMAT = "yyyy.MM.dd HH:mm:ss.SSSSSS";
- private static final KylinLogTool.ExtractLogByRangeTool LOG_EXTRACT = new KylinLogTool.ExtractLogByRangeTool(
- "^([0-9]{4}\\.[0-9]{2}\\.[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}\\.[0-9]{6})", SECOND_DATE_FORMAT);
- private String project;
-
- public ClickhouseDiagTool(String project) {
- this.project = project;
- }
-
- private static String getCompressedFileMatcher(String filePrefix, int maxCompressedFile) {
- Preconditions.checkState(maxCompressedFile > 0, "max file count should > 0," + maxCompressedFile);
-
- // /var/log/clickhouse-server/{*-server.log.0.gz,*-server.log.1.gz}
- return IntStream.range(0, maxCompressedFile).mapToObj(index -> filePrefix + "." + index + ".gz")
- .collect(Collectors.joining(",", "{", "}"));
- }
-
- private boolean isFullDiag() {
- return project == null;
- }
-
- public void dumpClickHouseServerLog(File exportDir, long startTime, long endTime) {
-
- try {
- SecondStorage.init(false);
-
- if (!SecondStorage.enabled()) {
- logger.error("TieredStorage is not enabled. Skip to fetch diag log.");
- return;
- }
-
- File ckLogsDir = new File(exportDir, SUB_DIR);
- FileUtils.forceMkdir(ckLogsDir);
- extractCkLogFile(ckLogsDir, startTime, endTime);
- } catch (Exception e) {
- logger.error("TieredStorage is enabled. But extract error,", e);
- }
-
- }
-
- private void extractCkLogFile(File exportDir, long startTime, long endTime) {
- val cluster = SecondStorage.configLoader().getCluster();
-
- if (cluster.emptyCluster()) {
- logger.error("TieredStorage cluster is empty. Skip to fetch diag log.");
- return;
- }
-
- if (StringUtils.isEmpty(cluster.getLogPath())) {
- logger.error("TieredStorage log path is empty. Skip to fetch diag log.");
- return;
- }
-
- val kylinConfig = KylinConfig.getInstanceFromEnv();
-
- String diagLogMatcher = kylinConfig.getSecondStorageDiagLogMatcher();
- File remoteFilePath = new File(cluster.getLogPath(), diagLogMatcher);
-
- int maxCompressedFile = kylinConfig.getSecondStorageDiagMaxCompressedFile();
- File remoteHistoryGZPath = maxCompressedFile > 0
- ? new File(cluster.getLogPath(), getCompressedFileMatcher(diagLogMatcher, maxCompressedFile))
- : null;
-
- val timeRange = new Pair<>(new DateTime(startTime).toString(SECOND_DATE_FORMAT),
- new DateTime(endTime).toString(SECOND_DATE_FORMAT));
-
- val allNodes = isFullDiag() ? cluster.getNodes() : SecondStorageNodeHelper.getALlNodesInProject(project);
-
- if (CollectionUtils.isEmpty(allNodes)) {
- logger.warn("There is no active node in TieredStorage");
- }
-
- allNodes.forEach(node -> {
- val cliCommandExecutor = new CliCommandExecutor(node.getIp(), cluster.getUserName(), cluster.getPassword(),
- kylinConfig.getSecondStorageSshIdentityPath(), node.getSSHPort());
- val nodeTargetPath = new File(exportDir,
- String.format(Locale.ROOT, CK_NODE_PATH_FORMAT, node.getName(), node.getIp(), node.getPort()));
- val nodeTargetTmpPath = new File(exportDir, nodeTargetPath.getName() + "_tmp");
-
- try {
- FileUtils.forceMkdir(nodeTargetPath);
- FileUtils.forceMkdir(nodeTargetTmpPath);
- cliCommandExecutor.copyRemoteToLocal(remoteFilePath.getAbsolutePath(),
- nodeTargetTmpPath.getAbsolutePath());
-
- copyAndUnzipCompressedLogFile(remoteHistoryGZPath, cliCommandExecutor, nodeTargetTmpPath);
-
- if (!extractCkLogByRange(timeRange, nodeTargetPath, nodeTargetTmpPath)) {
- return;
- }
-
- cleanEmptyFile(nodeTargetPath);
-
- } catch (IOException e) {
- logger.error("gather clickhouse log failed,{},", nodeTargetTmpPath.getAbsolutePath(), e);
- } finally {
- FileUtils.deleteQuietly(nodeTargetTmpPath);
- }
-
- });
- }
-
- private void copyAndUnzipCompressedLogFile(File remoteHistoryGZPath, CliCommandExecutor cliCommandExecutor,
- File nodeTargetTmpPath) {
- if (remoteHistoryGZPath == null) {
- return;
- }
-
- try {
- cliCommandExecutor.copyRemoteToLocal(remoteHistoryGZPath.getAbsolutePath(),
- nodeTargetTmpPath.getAbsolutePath());
- unzipLogFile(nodeTargetTmpPath);
- } catch (IOException e) {
- logger.error("copy remote compressed file failed,{},", nodeTargetTmpPath.getAbsolutePath(), e);
- }
-
- }
-
- private boolean extractCkLogByRange(Pair<String, String> timeRange, File nodeTargetPath, File nodeTargetTmpPath) {
- val serverLogFiles = nodeTargetTmpPath.listFiles((dir, name) -> name.endsWith(".log"));
-
- if (ArrayUtils.isEmpty(serverLogFiles)) {
- logger.error("{} dir is empty", nodeTargetTmpPath.getAbsolutePath());
- return false;
- }
-
- Arrays.stream(serverLogFiles).forEach(file -> {
- try {
- LOG_EXTRACT.extractLogByRange(file, timeRange, nodeTargetPath);
- } catch (IOException e) {
- logger.error("extract file:{} log error", file, e);
- }
- });
- return true;
- }
-
- private void cleanEmptyFile(File filePath) {
- val fileList = filePath.listFiles();
-
- if (ArrayUtils.isEmpty(fileList)) {
- return;
- }
-
- Arrays.stream(fileList).filter(file -> FileUtils.sizeOf(file) == 0).forEach(file -> {
-
- if (FileUtils.deleteQuietly(file)) {
- logger.debug("{} size is 0, clean it success", file.getName());
- return;
- }
- logger.debug("{} size is 0, clean it failed", file.getName());
- });
-
- }
-
- private void unzipLogFile(File nodeTargetPath) {
- val gzFileList = nodeTargetPath.listFiles((dir, name) -> name.endsWith(".gz"));
-
- if (ArrayUtils.isEmpty(gzFileList)) {
- logger.warn("{} dont have gz file, skit unzip it", nodeTargetPath.getName());
- return;
- }
-
- Arrays.stream(gzFileList).filter(file -> FileUtils.sizeOf(file) > 0).forEach(file -> {
- try {
-
- GZIPInputStream gzipInputStream = new GZIPInputStream(new FileInputStream(file));
- String fileName = file.getName().substring(0, file.getName().lastIndexOf("."));
- FileOutputStream fileOutputStream = new FileOutputStream(new File(nodeTargetPath, fileName + ".log"));
-
- byte[] buffer = new byte[1024];
- int len;
- while ((len = gzipInputStream.read(buffer)) > 0) {
- fileOutputStream.write(buffer, 0, len);
- }
-
- gzipInputStream.close();
- fileOutputStream.close();
-
- logger.info("Extracted " + fileName);
- } catch (IOException e) {
- logger.error("extract gz file:{} log error", file, e);
- }
- });
-
- }
-
-}
diff --git a/src/tool/src/test/java/org/apache/kylin/tool/ClickhouseDiagToolTest.java b/src/tool/src/test/java/org/apache/kylin/tool/ClickhouseDiagToolTest.java
deleted file mode 100644
index e9e14c66b9..0000000000
--- a/src/tool/src/test/java/org/apache/kylin/tool/ClickhouseDiagToolTest.java
+++ /dev/null
@@ -1,212 +0,0 @@
-/*
- * 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.kylin.tool;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.zip.GZIPOutputStream;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.filefilter.FileFilterUtils;
-import org.apache.kylin.common.util.Pair;
-import org.apache.kylin.common.util.NLocalFileMetadataTestCase;
-import org.joda.time.DateTime;
-import org.junit.Assert;
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.rules.TestName;
-import org.springframework.test.util.ReflectionTestUtils;
-
-import lombok.val;
-
-public class ClickhouseDiagToolTest extends NLocalFileMetadataTestCase {
-
- private static final String logs1 = "2020.05.18 18:15:50.745336 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 7 entries to flush\n"
- + "2020.05.18 18:15:50.760828 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 258.53 MiB.\n"
- + "2021.05.18 15:15:50.923367 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68538_68538_0 to 202105_87671_87671_0.\n"
- + "2021.05.18 15:15:50.924274 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:15:58.424386 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 8 entries to flush\n"
- + "2021.05.18 18:15:58.430833 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 539.80 MiB.\n"
- + "2021.05.18 18:15:58.434465 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68539_68539_0 to 202105_87672_87672_0.\n"
- + "2021.05.18 18:15:58.435039 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:05.935125 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 7 entries to flush\n"
- + "2021.05.18 18:16:05.940605 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 642.16 MiB.\n"
- + "2021.05.18 18:16:05.944313 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68540_68540_0 to 202105_87673_87673_0.\n"
- + "2021.05.18 18:16:05.945236 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:13.445337 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 8 entries to flush\n"
- + "2021.05.18 18:16:13.450534 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 876.80 MiB.\n"
- + "2021.05.18 18:16:13.453724 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68541_68541_0 to 202105_87674_87674_0.\n"
- + "2021.05.18 18:16:13.454482 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:20.954598 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 7 entries to flush\n"
- + "2021.05.18 18:16:20.960346 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 1.09 GiB.\n"
- + "2021.05.18 18:16:20.963832 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68542_68542_0 to 202105_87675_87675_0.\n"
- + "2021.05.18 18:16:20.964657 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2022.05.18 18:16:20.975851 [ 12151 ] {} <Debug> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71) (MergerMutator): Selected 2 parts from 202105_82621_87669_963 to 202105_87670_87670_0\n"
- + "2022.05.18 19:16:20.975914 [ 12151 ] {} <Debug> DiskLocal: Reserving 1.89 MiB on disk `default`, having unreserved 1.09 GiB.\n"
- + "2022.05.18 19:16:20.989143 [ 8600 ] {} <Debug> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71) (MergerMutator): Merging 2 parts: from 202105_82621_87669_963 to 202105_87670_8";
-
- private static final String logs2 = "2021.05.18 18:15:58.424386 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 8 entries to flush\n"
- + "2021.05.18 18:15:58.430833 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 539.80 MiB.\n"
- + "2021.05.18 18:15:58.434465 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68539_68539_0 to 202105_87672_87672_0.\n"
- + "2021.05.18 18:15:58.435039 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:05.935125 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 7 entries to flush\n"
- + "2021.05.18 18:16:05.940605 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 642.16 MiB.\n"
- + "2021.05.18 18:16:05.944313 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68540_68540_0 to 202105_87673_87673_0.\n"
- + "2021.05.18 18:16:05.945236 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:13.445337 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 8 entries to flush\n"
- + "2021.05.18 18:16:13.450534 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 876.80 MiB.\n"
- + "2021.05.18 18:16:13.453724 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68541_68541_0 to 202105_87674_87674_0.\n"
- + "2021.05.18 18:16:13.454482 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2021.05.18 18:16:20.954598 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushing system log, 7 entries to flush\n"
- + "2021.05.18 18:16:20.960346 [ 12121 ] {} <Debug> DiskLocal: Reserving 1.00 MiB on disk `default`, having unreserved 1.09 GiB.\n"
- + "2021.05.18 18:16:20.963832 [ 12121 ] {} <Trace> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71): Renaming temporary part tmp_insert_202105_68542_68542_0 to 202105_87675_87675_0.\n"
- + "2021.05.18 18:16:20.964657 [ 12121 ] {} <Trace> SystemLog (system.metric_log): Flushed system log\n"
- + "2022.05.18 18:16:20.975851 [ 12151 ] {} <Debug> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71) (MergerMutator): Selected 2 parts from 202105_82621_87669_963 to 202105_87670_87670_0\n"
- + "2022.05.18 19:16:20.975914 [ 12151 ] {} <Debug> DiskLocal: Reserving 1.89 MiB on disk `default`, having unreserved 1.09 GiB.\n"
- + "2022.05.18 19:16:20.989143 [ 8600 ] {} <Debug> system.metric_log (4ed24813-70a5-4891-a22a-5af9d7654f71) (MergerMutator): Merging 2 parts: from 202105_82621_87669_963 to 202105_87670_8\n";
-
- @Rule
- public TemporaryFolder temporaryFolder = new TemporaryFolder();
-
- @Rule
- public TestName testName = new TestName();
-
- @Test
- @Ignore
- public void testExtractCkLog() throws IOException {
- File mainDir = new File(temporaryFolder.getRoot(), testName.getMethodName());
-
- File subDir = new File(mainDir, ClickhouseDiagTool.SUB_DIR);
-
- File tmpDir = new File(subDir, "tmp");
- File targetDir = new File(subDir, "targetDir");
-
- FileUtils.forceMkdir(mainDir);
- FileUtils.forceMkdir(subDir);
- FileUtils.forceMkdir(tmpDir);
- FileUtils.forceMkdir(targetDir);
-
- File ckLog = new File(tmpDir, "click-server.log");
-
- val dataFormat = ReflectionTestUtils.getField(ClickhouseDiagTool.class, "SECOND_DATE_FORMAT").toString();
-
- //2021.05.18 18:15:00 - 2021.05.18 19:15:00
- val timeRange = new Pair<>(new DateTime(1621332900000L).toString(dataFormat),
- new DateTime(1621336500000L).toString(dataFormat));
-
- FileUtils.writeStringToFile(ckLog, logs1);
- Assert.assertTrue(ckLog.setLastModified(1621336500000L));
-
- val ckLogCount = Files.lines(Paths.get(ckLog.getAbsolutePath())).count();
-
- ClickhouseDiagTool clickhouseDiagTool = new ClickhouseDiagTool();
-
- ReflectionTestUtils.invokeMethod(clickhouseDiagTool, "extractCkLogByRange", timeRange, targetDir, tmpDir);
-
- File extractedLogFile = new File(targetDir, "click-server.log");
-
- val count = Files.lines(Paths.get(extractedLogFile.getAbsolutePath())).count();
- Assert.assertEquals(ckLogCount - count, 4);
-
- val extractedLogFileStr = FileUtils.readFileToString(extractedLogFile);
- Assert.assertEquals(extractedLogFileStr, logs2);
-
- }
-
- @Test
- public void testGenerateCompressedFile() {
- ClickhouseDiagTool clickhouseDiagTool = new ClickhouseDiagTool();
-
- {
- String serverLog = ReflectionTestUtils.invokeMethod(clickhouseDiagTool, "getCompressedFileMatcher",
- "*server-log.log", 2);
- Assert.assertEquals(serverLog, "{*server-log.log.0.gz,*server-log.log.1.gz}");
- }
-
- {
- String serverLog = ReflectionTestUtils.invokeMethod(clickhouseDiagTool, "getCompressedFileMatcher",
- "*server-log.log", 1);
- Assert.assertEquals(serverLog, "{*server-log.log.0.gz}");
- }
-
- }
-
- @Test
- public void testCleanEmptyFile() throws IOException {
- File mainDir = new File(temporaryFolder.getRoot(), testName.getMethodName());
- FileUtils.forceMkdir(mainDir);
- FileUtils.touch(new File(mainDir, "click-server.log"));
- val fileFilter = FileFilterUtils.suffixFileFilter("log");
-
- val fileList = FileUtils.listFiles(mainDir, fileFilter, null);
-
- Assert.assertEquals(1, fileList.size());
- fileList.forEach(file -> {
- Assert.assertEquals(0, FileUtils.sizeOf(file));
- });
-
- ClickhouseDiagTool clickhouseDiagTool = new ClickhouseDiagTool();
-
- ReflectionTestUtils.invokeMethod(clickhouseDiagTool, "cleanEmptyFile", mainDir);
-
- Assert.assertEquals(0, FileUtils.listFiles(mainDir, fileFilter, null).size());
-
- }
-
- @Test
- public void testUnzipLogFile() throws IOException {
- File mainDir = new File(temporaryFolder.getRoot(), testName.getMethodName());
- val fileFilter = FileFilterUtils.suffixFileFilter("log");
- File tarTempDir = new File(mainDir, "tarTemp");
- FileUtils.forceMkdir(mainDir);
- FileUtils.forceMkdir(tarTempDir);
- val logFile = new File(mainDir, "click-server.log");
- FileUtils.touch(logFile);
- FileUtils.writeStringToFile(logFile, "abc");
-
- writeFileToTar(new File(tarTempDir, "t.gz"), logFile);
-
- ClickhouseDiagTool clickhouseDiagTool = new ClickhouseDiagTool();
- ReflectionTestUtils.invokeMethod(clickhouseDiagTool, "unzipLogFile", tarTempDir);
-
- Assert.assertEquals(1, FileUtils.listFiles(tarTempDir, fileFilter, null).size());
-
- }
-
- private void writeFileToTar(File tarTempFile, File logFile) throws IOException {
- GZIPOutputStream out = new GZIPOutputStream(new FileOutputStream(tarTempFile));
- InputStream in = new FileInputStream(logFile);
-
- byte[] buf = new byte[1024];
- int len;
- while ((len = in.read(buf)) > 0) {
- out.write(buf, 0, len);
- }
- in.close();
-
- out.finish();
- out.close();
- }
-}