You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by jl...@apache.org on 2025/09/15 02:56:08 UTC
(kylin) branch kylin5 updated: KYLIN-6078 submit jobs when refresh specify partitions
This is an automated email from the ASF dual-hosted git repository.
jlfsdtc pushed a commit to branch kylin5
in repository https://gitbox.apache.org/repos/asf/kylin.git
The following commit(s) were added to refs/heads/kylin5 by this push:
new abdc910760 KYLIN-6078 submit jobs when refresh specify partitions
abdc910760 is described below
commit abdc9107606844e56d1581079a0ec3a62f04e953
Author: jlf <12...@qq.com>
AuthorDate: Mon Sep 15 10:56:03 2025 +0800
KYLIN-6078 submit jobs when refresh specify partitions
---
.../rest/request/InternalTableBuildRequest.java | 2 +-
.../kylin/job/handler/InternalTableJobHandler.java | 2 +
.../kylin/metadata/cube/model/NBatchConstants.java | 8 +-
.../java/org/apache/kylin/util/DataRangeUtils.java | 8 +-
.../kylin/rest/response/ExecutableResponse.java | 15 ++--
.../rest/service/InternalTableServiceTest.java | 13 ++-
.../apache/kylin/rest/service/JobServiceTest.java | 5 --
.../rest/service/InternalTableLoadingService.java | 93 ++++++++++++----------
.../kylin/rest/service/InternalTableService.java | 3 +
.../engine/spark/job/InternalTableLoadJob.java | 8 +-
.../engine/spark/job/InternalTableLoadingJob.java | 23 ++----
.../spark/job/InternalTableUpdateMetadataStep.java | 51 ++++++++----
.../engine/spark/builder/InternalTableLoader.scala | 5 +-
13 files changed, 141 insertions(+), 95 deletions(-)
diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/request/InternalTableBuildRequest.java b/src/common-service/src/main/java/org/apache/kylin/rest/request/InternalTableBuildRequest.java
index 434f1bdbaf..49359e817c 100644
--- a/src/common-service/src/main/java/org/apache/kylin/rest/request/InternalTableBuildRequest.java
+++ b/src/common-service/src/main/java/org/apache/kylin/rest/request/InternalTableBuildRequest.java
@@ -41,6 +41,6 @@ public class InternalTableBuildRequest {
private String yarnQueue;
@JsonProperty("partitions")
- private String[] partitions;
+ private String[] partitions = new String[] {};
}
diff --git a/src/core-job/src/main/java/org/apache/kylin/job/handler/InternalTableJobHandler.java b/src/core-job/src/main/java/org/apache/kylin/job/handler/InternalTableJobHandler.java
index d127ccb76c..908f4f8c75 100644
--- a/src/core-job/src/main/java/org/apache/kylin/job/handler/InternalTableJobHandler.java
+++ b/src/core-job/src/main/java/org/apache/kylin/job/handler/InternalTableJobHandler.java
@@ -39,6 +39,7 @@ public class InternalTableJobHandler extends AbstractJobHandler {
private String endDate;
private String deletePartitionValues;
private String deletePartition; // true or false
+ private String refreshPartitionValues;
public InternalTableJobBuildParam(JobParam jobParam) {
super(null, null, jobParam.getOwner(), jobParam.getJobTypeEnum(), jobParam.getJobId(), null, null, null,
@@ -51,6 +52,7 @@ public class InternalTableJobHandler extends AbstractJobHandler {
this.endDate = jobParam.getExtParams().get(NBatchConstants.P_END_DATE);
this.deletePartitionValues = jobParam.getExtParams().get(NBatchConstants.P_DELETE_PARTITION_VALUES);
this.deletePartition = jobParam.getExtParams().get(NBatchConstants.P_DELETE_PARTITION);
+ this.refreshPartitionValues = jobParam.getExtParams().get(NBatchConstants.P_REFRESH_PARTITION_VALUES);
}
}
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/metadata/cube/model/NBatchConstants.java b/src/core-metadata/src/main/java/org/apache/kylin/metadata/cube/model/NBatchConstants.java
index fab3fab21a..ea34305552 100644
--- a/src/core-metadata/src/main/java/org/apache/kylin/metadata/cube/model/NBatchConstants.java
+++ b/src/core-metadata/src/main/java/org/apache/kylin/metadata/cube/model/NBatchConstants.java
@@ -76,7 +76,7 @@ public interface NBatchConstants {
String P_DELETE_PARTITION = "deletePartition";
String P_SORT_BY_PARTITION_BEFORE_SAVE = "sortByPartition";
String P_PRELOADED_CACHE = "preloadedCache";
-
+ String P_REFRESH_PARTITION_VALUES = "refreshPartitionValues";
/** index planner job parameters */
String P_PLANNER_INITIALIZE_CUBOID_COUNT = "kylin.planner.initializeCuboidCount";
@@ -92,10 +92,8 @@ public interface NBatchConstants {
@Getter
enum TblPropertyKey {
- PRIMARY_KEY(P_PRIMARY_KEY),
- ORDER_BY_KEY(P_ORDER_BY_KEY),
- BUCKET_COLUMN(P_BUCKET_COLUMN),
- BUCKET_NUM(P_BUCKET_NUM);
+ PRIMARY_KEY(P_PRIMARY_KEY), ORDER_BY_KEY(P_ORDER_BY_KEY), BUCKET_COLUMN(P_BUCKET_COLUMN), BUCKET_NUM(
+ P_BUCKET_NUM);
private final String value;
diff --git a/src/core-metadata/src/main/java/org/apache/kylin/util/DataRangeUtils.java b/src/core-metadata/src/main/java/org/apache/kylin/util/DataRangeUtils.java
index ccbc6556ce..f23cba0205 100644
--- a/src/core-metadata/src/main/java/org/apache/kylin/util/DataRangeUtils.java
+++ b/src/core-metadata/src/main/java/org/apache/kylin/util/DataRangeUtils.java
@@ -35,6 +35,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.exception.KylinException;
import org.apache.kylin.common.util.DateFormat;
import org.apache.kylin.guava30.shaded.common.base.Preconditions;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.metadata.model.PartitionDesc;
public final class DataRangeUtils {
@@ -169,13 +170,18 @@ public final class DataRangeUtils {
if (values == null || values.isEmpty()) {
return mergedRanges;
}
-
SimpleDateFormat sdf = new SimpleDateFormat(dateFormat, Locale.ROOT);
try {
List<Date> dates = new ArrayList<>();
for (String value : values) {
+ if (StringUtils.isEmpty(value)) {
+ continue;
+ }
dates.add(sdf.parse(value));
}
+ if (dates.isEmpty()) {
+ return Lists.newArrayList();
+ }
// Sort the dates
dates.sort(Date::compareTo);
Date start = dates.get(0);
diff --git a/src/data-loading-service/src/main/java/org/apache/kylin/rest/response/ExecutableResponse.java b/src/data-loading-service/src/main/java/org/apache/kylin/rest/response/ExecutableResponse.java
index ddf6635482..24080785fe 100644
--- a/src/data-loading-service/src/main/java/org/apache/kylin/rest/response/ExecutableResponse.java
+++ b/src/data-loading-service/src/main/java/org/apache/kylin/rest/response/ExecutableResponse.java
@@ -84,6 +84,8 @@ public class ExecutableResponse implements Comparable<ExecutableResponse> {
private ExecutableState schedulerState;
@JsonProperty("job_name")
private String jobName;
+ @JsonProperty("data_range_partitions")
+ private String dataRangePartitions;
@JsonProperty("data_range_start")
private long dataRangeStart;
@JsonProperty("data_range_end")
@@ -202,9 +204,12 @@ public class ExecutableResponse implements Comparable<ExecutableResponse> {
}
} else if (abstractExecutable instanceof InternalTableLoadingJob) {
InternalTableLoadingJob internalTableJob = (InternalTableLoadingJob) abstractExecutable;
- if ("false".equals(internalTableJob.getParam("incrementalBuild"))
- || "true".equals(internalTableJob.getParam("deletePartition"))) {
+ String partitionValues = internalTableJob.getParam(NBatchConstants.P_REFRESH_PARTITION_VALUES);
+ if ("false".equals(internalTableJob.getParam("incrementalBuild"))) {
executableResponse.setDataRangeEnd(Long.MAX_VALUE);
+ } else if (StringUtils.isNotEmpty(partitionValues)) {
+ partitionValues = partitionValues.replace("[", "").replace("]", "");
+ executableResponse.setDataRangePartitions(partitionValues);
} else {
executableResponse.setDataRangeStart(Long.parseLong(internalTableJob.getParam("startTime")));
executableResponse.setDataRangeEnd(Long.parseLong(internalTableJob.getParam("endTime")));
@@ -288,8 +293,7 @@ public class ExecutableResponse implements Comparable<ExecutableResponse> {
/** calculate stage count from segment */
public static double calculateSuccessStageInTaskMap(AbstractExecutable task,
- Map<String, List<StageExecutable>> stageMap,
- ExecutablePO executablePO) {
+ Map<String, List<StageExecutable>> stageMap, ExecutablePO executablePO) {
var successStages = 0D;
boolean calculateIndexExecRadio = stageMap.size() == 1;
for (Map.Entry<String, List<StageExecutable>> entry : stageMap.entrySet()) {
@@ -301,8 +305,7 @@ public class ExecutableResponse implements Comparable<ExecutableResponse> {
}
public static double calculateSuccessStage(AbstractExecutable task, String segmentId,
- List<StageExecutable> stageExecutables,
- boolean calculateIndexExecRadio, ExecutablePO executablePO) {
+ List<StageExecutable> stageExecutables, boolean calculateIndexExecRadio, ExecutablePO executablePO) {
var successStages = 0D;
for (StageExecutable stage : stageExecutables) {
if (ExecutableState.SUCCEED == stage.getStatusInMem(segmentId)
diff --git a/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/InternalTableServiceTest.java b/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/InternalTableServiceTest.java
index 3cda8cac17..7b3aca4def 100644
--- a/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/InternalTableServiceTest.java
+++ b/src/data-loading-service/src/test/java/org/apache/kylin/rest/service/InternalTableServiceTest.java
@@ -551,6 +551,13 @@ public class InternalTableServiceTest extends AbstractTestCase {
ExecutableState state = executableManager.getJob(finalJobId3).getStatus();
return state.isFinalState() || state == ExecutableState.ERROR;
});
+
+ // test cancel a full load job
+ internalTableService.truncateInternalTable(PROJECT, TABLE_INDENTITY);
+ jobId = internalTableService
+ .loadIntoInternalTable(PROJECT, table.getName(), table.getDatabase(), false, false, "", "", null, null)
+ .getJobs().get(0).getJobId();
+ executableManager.discardJob(jobId);
}
@Test
@@ -578,6 +585,7 @@ public class InternalTableServiceTest extends AbstractTestCase {
void testRefreshPartitions() throws Exception {
KylinConfig config = KylinConfig.getInstanceFromEnv();
NTableMetadataManager tManager = NTableMetadataManager.getInstance(config, PROJECT);
+ ExecutableManager executableManager = ExecutableManager.getInstance(config, PROJECT);
TableDesc table = tManager.getTableDesc(TABLE_INDENTITY);
internalTableService.createInternalTable(PROJECT, TABLE_INDENTITY, new String[] { PARTITION_COL }, null,
new HashMap<>(), InternalTableDesc.StorageType.PARQUET.name());
@@ -593,8 +601,9 @@ public class InternalTableServiceTest extends AbstractTestCase {
when(tableService.getPartitionColumnFormat(any(), any(), any(), any())).thenReturn("yyyyMM");
internalTableService.updateInternalTable(PROJECT, table.getName(), table.getDatabase(),
new String[] { DATE_COL }, "yyyyMM", new HashMap<>(), InternalTableDesc.StorageType.PARQUET.name());
- internalTableService.loadIntoInternalTable(PROJECT, table.getName(), table.getDatabase(), false, true, "", "",
- new String[] { "199201", "199203" }, null);
+ String jobId = internalTableService.loadIntoInternalTable(PROJECT, table.getName(), table.getDatabase(), false,
+ true, "", "", new String[] { "199201", "199203" }, null).getJobs().get(0).getJobId();
+ executableManager.discardJob(jobId);
}
@Test
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 a697d9c66a..f0f6a8ded5 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
@@ -1009,11 +1009,6 @@ public class JobServiceTest extends NLocalFileMetadataTestCase {
response = ExecutableResponse.create(job, null);
Assert.assertEquals(Long.MAX_VALUE, response.getDataRangeEnd());
- job.setParam("incrementalBuild", "true");
- job.setParam("deletePartition", "true");
- response = ExecutableResponse.create(job, null);
- Assert.assertEquals(Long.MAX_VALUE, response.getDataRangeEnd());
-
TableDesc originTable = manager.getTableDesc("DEFAULT.TEST_KYLIN_FACT");
internalManager.createInternalTable(new InternalTableDesc(originTable));
diff --git a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableLoadingService.java b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableLoadingService.java
index ef42c303b1..884f26bb70 100644
--- a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableLoadingService.java
+++ b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableLoadingService.java
@@ -27,6 +27,8 @@ import static org.apache.kylin.job.execution.JobTypeEnum.INTERNAL_TABLE_REFRESH;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.Objects;
@@ -39,6 +41,7 @@ import org.apache.kylin.common.util.TimeUtil;
import org.apache.kylin.engine.spark.builder.InternalTableLoader;
import org.apache.kylin.engine.spark.job.InternalTableLoadJob;
import org.apache.kylin.engine.spark.job.InternalTableUpdateMetadataStep;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
import org.apache.kylin.job.dao.JobStatisticsManager;
import org.apache.kylin.job.execution.JobTypeEnum;
@@ -75,28 +78,22 @@ public class InternalTableLoadingService extends BasicService {
jobStatisticsManager.updateStatistics(TimeUtil.getDayStart(System.currentTimeMillis()), 0, 0, 1);
InternalTableDesc internalTable = checkAndGetInternalTables(project, table, database);
InternalTableManager internalTableManager = InternalTableManager.getInstance(getConfig(), project);
- // refresh partitions
- if (isRefresh && null != partitions && partitions.length > 0) {
- SparkSession ss = SparkSession.getDefaultSession().get();
- InternalTableLoader internalTableLoader = new InternalTableLoader();
- internalTableLoader.loadInternalTable(ss, internalTable, new String[] { startDate, endDate },
- partitions, internalTable.getStorageType().getFormat(), isIncremental);
- } else {
- checkBeforeSubmit(internalTable, jobType, isIncremental, isRefresh, startDate, endDate);
- logger.info(
- "create internal table loading job for table: {}, isIncrementBuild: {}, startTime: {}, endTime: {}",
- internalTable.getIdentity(), isIncremental, startDate, endDate);
- JobParam jobParam = new JobParam().withProject(project).withTable(internalTable.getIdentity())
- .withYarnQueue(yarnQueue).withJobTypeEnum(jobType).withOwner(BasicService.getUsername())
- .addExtParams(NBatchConstants.P_INCREMENTAL_BUILD, String.valueOf(isIncremental))
- .addExtParams(NBatchConstants.P_OUTPUT_MODE, String.valueOf(isRefresh))
- .addExtParams(NBatchConstants.P_START_DATE, startDate)
- .addExtParams(NBatchConstants.P_END_DATE, endDate);
- String jobId = getManager(SourceUsageManager.class).licenseCheckWrap(project,
- () -> getManager(JobManager.class, project).addJob(jobParam));
- jobIds.add(jobId);
- internalTableManager.saveOrUpdateInternalTable(internalTable);
- }
+ checkBeforeSubmit(internalTable, isIncremental, isRefresh, startDate, endDate, partitions);
+ logger.info(
+ "create internal table loading job for table: {}, isIncrementBuild: {}, startTime: {}, endTime: {}",
+ internalTable.getIdentity(), isIncremental, startDate, endDate);
+ String partitionValues = null == partitions ? "" : Arrays.toString(partitions);
+ JobParam jobParam = new JobParam().withProject(project).withTable(internalTable.getIdentity())
+ .withYarnQueue(yarnQueue).withJobTypeEnum(jobType).withOwner(BasicService.getUsername())
+ .addExtParams(NBatchConstants.P_INCREMENTAL_BUILD, String.valueOf(isIncremental))
+ .addExtParams(NBatchConstants.P_OUTPUT_MODE, String.valueOf(isRefresh))
+ .addExtParams(NBatchConstants.P_START_DATE, startDate)
+ .addExtParams(NBatchConstants.P_END_DATE, endDate)
+ .addExtParams(NBatchConstants.P_REFRESH_PARTITION_VALUES, partitionValues);
+ String jobId = getManager(SourceUsageManager.class).licenseCheckWrap(project,
+ () -> getManager(JobManager.class, project).addJob(jobParam));
+ jobIds.add(jobId);
+ internalTableManager.saveOrUpdateInternalTable(internalTable);
return true;
}, project);
String jobName = isRefresh ? INTERNAL_TABLE_REFRESH.toString() : INTERNAL_TABLE_BUILD.toString();
@@ -108,42 +105,53 @@ public class InternalTableLoadingService extends BasicService {
* @param startDate
* @param endDate
*/
- private void checkBeforeSubmit(InternalTableDesc internalTable, JobTypeEnum jobType, boolean isIncremental,
- boolean isRefresh, String startDate, String endDate) throws KylinException {
+ private void checkBeforeSubmit(InternalTableDesc internalTable, boolean isIncremental, boolean isRefresh,
+ String startDate, String endDate, String[] partitions) throws Exception {
if (isIncremental && (Objects.isNull(internalTable.getTablePartition())
|| Objects.isNull(internalTable.getTablePartition().getPartitionColumns())
|| internalTable.getTablePartition().getPartitionColumns().length == 0)) {
String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getInternalTableUnpartitioned());
throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
}
+ partitions = null == partitions ? new String[] {} : partitions;
// check job_range overlap?
InternalTablePartition tablePartition = internalTable.getTablePartition();
List<String[]> jobRange = internalTable.getJobRange();
- String[] curJobRange = new String[] { "0", "0" };
+ List<String[]> curRange = Lists.newArrayList();
+ if (!isIncremental) {
+ curRange.add(new String[] { "0", "0" });
+ }
String timeFmt = Objects.isNull(tablePartition) ? "" : tablePartition.getDatePartitionFormat();
+ if (isRefresh && partitions.length > 0 && StringUtils.isNotEmpty(timeFmt)) {
+ curRange.addAll(DataRangeUtils.mergeTimeRange(Arrays.asList(partitions), timeFmt));
+ }
if (StringUtils.isNotEmpty(startDate) && StringUtils.isNotEmpty(timeFmt)) {
SimpleDateFormat fmt = new SimpleDateFormat(timeFmt, Locale.ROOT);
- String start = StringUtils.isEmpty(startDate) ? "0" : fmt.format(Long.parseLong(startDate));
- String end = StringUtils.isEmpty(endDate) ? "0" : fmt.format(Long.parseLong(endDate));
- curJobRange = new String[] { start, end };
+ String start = fmt.format(Long.parseLong(startDate));
+ String end = fmt.format(Long.parseLong(endDate));
+ curRange.add(new String[] { start, end });
}
// non-time partition table can not submit incremental job
- if (isIncremental && !Objects.isNull(tablePartition)
+ if (isIncremental && !Objects.isNull(tablePartition) && StringUtils.isNotEmpty(startDate)
&& StringUtils.isEmpty(tablePartition.getDatePartitionFormat())) {
String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getNonTimeInternalTableIncrementalBuild());
throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
}
- if (DataRangeUtils.timeOverlap(internalTable.getJobRange(), curJobRange, timeFmt)) {
- String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getTimeRangeOverlap());
- throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
- }
- // check refresh out of data range
- if (isRefresh && !DataRangeUtils.timeInRange(curJobRange, internalTable.getPartitionRange(), timeFmt)) {
- String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getTimeOutOfRange());
- throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
- }
- logger.info(jobType.getCategory());
- jobRange.add(curJobRange);
+ String[] finalPartitions = partitions;
+ curRange.forEach(range -> {
+ if (DataRangeUtils.timeOverlap(internalTable.getJobRange(), range, timeFmt)) {
+ String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getTimeRangeOverlap());
+ throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
+ }
+ // check refresh out of data range(exclude specify partitions)
+ if (isRefresh && finalPartitions.length == 0
+ && !DataRangeUtils.timeInRange(range, internalTable.getPartitionRange(), timeFmt)) {
+ String errorMsg = String.format(Locale.ROOT, MsgPicker.getMsg().getInternalTableUnpartitioned());
+ throw new KylinException(INTERNAL_TABLE_ERROR, errorMsg);
+ }
+ });
+ jobRange.addAll(curRange);
+ jobRange.sort(Comparator.comparing(valueA -> valueA[0]));
internalTable.setJobRange(jobRange);
}
@@ -168,6 +176,11 @@ public class InternalTableLoadingService extends BasicService {
tablePartition.setPartitionValues(info.getPartitionValues());
tablePartition.setPartitionDetails(info.getPartitionDetails());
oldTable.setRowCount(info.getFinalCount());
+ if (StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())) {
+ List<String[]> partitionRange = DataRangeUtils.mergeTimeRange(tablePartition.getPartitionValues(),
+ tablePartition.getDatePartitionFormat());
+ oldTable.setPartitionRange(partitionRange);
+ }
internalTableManager.saveOrUpdateInternalTable(oldTable);
return true;
}, project);
diff --git a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableService.java b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableService.java
index 4496ac5a0f..33520a4e86 100644
--- a/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableService.java
+++ b/src/datasource-service/src/main/java/org/apache/kylin/rest/service/InternalTableService.java
@@ -406,6 +406,9 @@ public class InternalTableService extends BasicService {
if (isIncremental) {
DataRangeUtils.validateRange(startDate, endDate);
}
+ if (null != partitions && partitions.length > 0) {
+ isIncremental = true;
+ }
// treat full refresh as full load processing.
if (!isIncremental && isRefresh) {
isRefresh = false;
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadJob.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadJob.java
index 10f8b4562e..a6347ae760 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadJob.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadJob.java
@@ -72,9 +72,15 @@ public class InternalTableLoadJob extends SparkApplication {
boolean incrementalBuild = "true".equals(getParam(NBatchConstants.P_INCREMENTAL_BUILD));
String startDate = getParam(NBatchConstants.P_START_DATE);
String endDate = getParam(NBatchConstants.P_END_DATE);
+ String refreshPartitions = getParam(NBatchConstants.P_REFRESH_PARTITION_VALUES).replace("[", "").replace("]",
+ "");
+ String[] partitions = new String[] {};
+ if (StringUtils.isNotEmpty(refreshPartitions)) {
+ partitions = StringUtils.isEmpty(refreshPartitions) ? new String[] {} : refreshPartitions.split(", ");
+ }
String storagePolicy = config.getGlutenStoragePolicy();
InternalTableLoader loader = new InternalTableLoader();
- loader.loadInternalTable(ss, internalTable, new String[] { startDate, endDate }, null, storagePolicy,
+ loader.loadInternalTable(ss, internalTable, new String[] { startDate, endDate }, partitions, storagePolicy,
incrementalBuild);
}
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadingJob.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadingJob.java
index efb331e09a..c73a9d7f44 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadingJob.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableLoadingJob.java
@@ -18,9 +18,7 @@
package org.apache.kylin.engine.spark.job;
-import java.text.SimpleDateFormat;
import java.util.List;
-import java.util.Locale;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
@@ -74,6 +72,7 @@ public class InternalTableLoadingJob extends DefaultExecutableOnTable {
job.setParam(NBatchConstants.P_END_DATE, param.getEndDate());
job.setParam(NBatchConstants.P_DELETE_PARTITION_VALUES, param.getDeletePartitionValues());
job.setParam(NBatchConstants.P_DELETE_PARTITION, param.getDeletePartition());
+ job.setParam(NBatchConstants.P_REFRESH_PARTITION_VALUES, param.getRefreshPartitionValues());
KylinConfig config = KylinConfig.getInstanceFromEnv();
StepEnum.BUILD_INTERNAL.create(job, config);
StepEnum.UPDATE_METADATA.create(job, config);
@@ -114,15 +113,10 @@ public class InternalTableLoadingJob extends DefaultExecutableOnTable {
String tableName = getParam(NBatchConstants.P_TABLE_NAME);
String startDate = getParam(NBatchConstants.P_START_DATE);
String endDate = getParam(NBatchConstants.P_END_DATE);
+ String refreshPartitions = getParam(NBatchConstants.P_REFRESH_PARTITION_VALUES).replace("[", "")
+ .replace("]", "");
+ boolean isIncremental = "true".equals(getParam(NBatchConstants.P_INCREMENTAL_BUILD));
InternalTablePartition tablePartition = internalTable.getTablePartition();
- // release current job_range
- String[] curJobRange = new String[] { "0", "0" };
- if (null != tablePartition && StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())
- && StringUtils.isNotEmpty(startDate)) {
- SimpleDateFormat fmt = new SimpleDateFormat(tablePartition.getDatePartitionFormat(), Locale.ROOT);
- curJobRange = new String[] { fmt.format(Long.parseLong(startDate)),
- fmt.format(Long.parseLong(endDate)) };
- }
// merge latest partition_range
logger.info("starting merging delta partitions for internal table {}", tableName);
if (null != tablePartition && StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())) {
@@ -130,12 +124,11 @@ public class InternalTableLoadingJob extends DefaultExecutableOnTable {
tablePartition.getDatePartitionFormat());
internalTable.setPartitionRange(partitionRange);
}
- List<String[]> jobRange = internalTable.getJobRange();
- String[] finalCurJobRange = curJobRange;
- jobRange.removeIf(rang -> rang[0].equals(finalCurJobRange[0]) && rang[1].equals(finalCurJobRange[1]));
- internalTable.setJobRange(jobRange);
+ // release current job_range
+ InternalTableUpdateMetadataStep metadataStep = new InternalTableUpdateMetadataStep();
+ metadataStep.releaseJobRange(internalTable, isIncremental, startDate, endDate, refreshPartitions);
internalTableManager.saveOrUpdateInternalTable(internalTable);
- logger.info("release job_range for internal table {} , range {}.", tableName, jobRange);
+ logger.info("release job_range for internal table {} ", tableName);
return true;
}, getProject());
}
diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableUpdateMetadataStep.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableUpdateMetadataStep.java
index 46c3c2065a..abf20ba3a7 100644
--- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableUpdateMetadataStep.java
+++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/InternalTableUpdateMetadataStep.java
@@ -21,6 +21,7 @@ package org.apache.kylin.engine.spark.job;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -36,6 +37,7 @@ import org.apache.kylin.common.util.HadoopUtil;
import org.apache.kylin.engine.spark.builder.InternalTableLoader;
import org.apache.kylin.guava30.shaded.common.annotations.VisibleForTesting;
import org.apache.kylin.guava30.shaded.common.base.Preconditions;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.job.JobContext;
import org.apache.kylin.job.constant.ExecutableConstants;
import org.apache.kylin.job.exception.ExecuteException;
@@ -92,6 +94,9 @@ public class InternalTableUpdateMetadataStep extends AbstractExecutable {
String project = getParam(NBatchConstants.P_PROJECT_NAME);
String startDate = getParam(NBatchConstants.P_START_DATE);
String endDate = getParam(NBatchConstants.P_END_DATE);
+ String refreshPartitions = getParam(NBatchConstants.P_REFRESH_PARTITION_VALUES).replace("[", "")
+ .replace("]", "");
+ boolean isIncremental = "true".equals(getParam(NBatchConstants.P_INCREMENTAL_BUILD));
// fetch delta partition info
InternalTableManager internalTableManager = InternalTableManager.getInstance(config, project);
InternalTableDesc internalTable = internalTableManager.getInternalTableDesc(tableName);
@@ -109,21 +114,7 @@ public class InternalTableUpdateMetadataStep extends AbstractExecutable {
tablePartition.getDatePartitionFormat());
internalTable.setPartitionRange(partitionRange);
}
- // release current job_range
- String[] curJobRange = new String[] { "0", "0" };
- if (null != tablePartition && StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())
- && StringUtils.isNotEmpty(startDate)) {
- SimpleDateFormat fmt = new SimpleDateFormat(tablePartition.getDatePartitionFormat(), Locale.ROOT);
- curJobRange = new String[] { fmt.format(Long.parseLong(startDate)),
- fmt.format(Long.parseLong(endDate)) };
- }
- List<String[]> jobRange = internalTable.getJobRange();
- String[] finalCurJobRange = curJobRange;
- jobRange.removeIf(rang -> rang[0].equals(finalCurJobRange[0]) && rang[1].equals(finalCurJobRange[1]));
- internalTable.setJobRange(jobRange);
- logger.info("trying to release job_range for internal table {} , range {}.",
- internalTable.getTableDesc().getTableAlias(), jobRange);
-
+ releaseJobRange(internalTable, isIncremental, startDate, endDate, refreshPartitions);
internalTableManager.saveOrUpdateInternalTable(internalTable);
logger.info("update metadata for internal table {} cost: {} ms.",
internalTable.getTableDesc().getTableAlias(), (System.currentTimeMillis() - startTime));
@@ -131,6 +122,36 @@ public class InternalTableUpdateMetadataStep extends AbstractExecutable {
}, project);
}
+ public void releaseJobRange(InternalTableDesc internalTable, boolean isIncremental, String startDate,
+ String endDate, String refreshPartitions) throws Exception {
+ InternalTablePartition tablePartition = internalTable.getTablePartition();
+ // release current job_range
+ logger.info("starting releasing job_range");
+ List<String[]> curRange = Lists.newArrayList();
+ if (!isIncremental) {
+ curRange.add(new String[] { "0", "0" });
+ }
+ if (StringUtils.isNotEmpty(refreshPartitions)
+ && StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())) {
+ String[] partitions = refreshPartitions.split(", ");
+ curRange.addAll(
+ DataRangeUtils.mergeTimeRange(Arrays.asList(partitions), tablePartition.getDatePartitionFormat()));
+ }
+ if (null != tablePartition && StringUtils.isNotEmpty(tablePartition.getDatePartitionFormat())
+ && StringUtils.isNotEmpty(startDate)) {
+ SimpleDateFormat fmt = new SimpleDateFormat(tablePartition.getDatePartitionFormat(), Locale.ROOT);
+ curRange.add(new String[] { fmt.format(Long.parseLong(startDate)), fmt.format(Long.parseLong(endDate)) });
+ }
+ List<String[]> jobRange = internalTable.getJobRange();
+ curRange.forEach(curJobRange -> {
+ jobRange.removeIf(range -> range[0].equals(curJobRange[0]) && range[1].equals(curJobRange[1]));
+ });
+ internalTable.setJobRange(jobRange);
+ logger.info("trying to release job_range for internal table {} , range {}.",
+ internalTable.getTableDesc().getTableAlias(), jobRange);
+
+ }
+
public InternalTableLoadJob.InternalTableMetaUpdateInfo extractUpdateInfo(String project, String tableName,
KylinConfig config, SparkSession ss) {
InternalTableManager internalTableManager = InternalTableManager.getInstance(config, project);
diff --git a/src/spark-project/engine-spark/src/main/scala/org/apache/kylin/engine/spark/builder/InternalTableLoader.scala b/src/spark-project/engine-spark/src/main/scala/org/apache/kylin/engine/spark/builder/InternalTableLoader.scala
index 218b672dbf..d2927a63e3 100644
--- a/src/spark-project/engine-spark/src/main/scala/org/apache/kylin/engine/spark/builder/InternalTableLoader.scala
+++ b/src/spark-project/engine-spark/src/main/scala/org/apache/kylin/engine/spark/builder/InternalTableLoader.scala
@@ -25,7 +25,7 @@ import java.util.Locale
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.fs.Path
import org.apache.kylin.common.exception.{CommonErrorCode, KylinException}
-import org.apache.kylin.common.util.{DateFormat, HadoopUtil}
+import org.apache.kylin.common.util.HadoopUtil
import org.apache.kylin.engine.spark.utils.SparkDataSource._
import org.apache.kylin.metadata.cube.model.NBatchConstants
import org.apache.kylin.metadata.table.InternalTableDesc
@@ -102,9 +102,6 @@ class InternalTableLoader extends Logging {
}
val format = table.getStorageType.getFormat
if (incremental) {
- val dateFormat = table.getTablePartition.getDatePartitionFormat
- logInfo(f"Refresh dynamic partitions [${DateFormat.formatToDateStr(startDate.toLong, dateFormat)}," +
- f" ${DateFormat.formatToDateStr(endDate.toLong, dateFormat)})")
ss.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
}
writer.format(format).mode(outPutMode).save(location)