weizhengte commented on code in PR #8859:
URL: https://github.com/apache/incubator-doris/pull/8859#discussion_r843929072
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobScheduler.java:
##########
@@ -33,31 +49,217 @@
Switch job state from pending to scheduling.
*/
public class StatisticsJobScheduler extends MasterDaemon {
+ private static final Logger LOG =
LogManager.getLogger(StatisticsJobScheduler.class);
- public Queue<StatisticsJob> pendingJobQueue =
Queues.newLinkedBlockingQueue();
+ /**
+ * Different statistics need to be collected for the jobs submitted by
users.
+ * if all statistics be collected at the same time, the cluster may be
overburdened
+ * and normal query services may be affected. Therefore, we put the jobs
into the queue
+ * and schedule them one by one, and finally divide each job to several
subtasks and execute them.
+ */
+ public final Queue<StatisticsJob> pendingJobQueue =
Queues.newLinkedBlockingQueue(Config.cbo_max_statistics_job_num);
public StatisticsJobScheduler() {
super("Statistics job scheduler", 0);
}
@Override
protected void runAfterCatalogReady() {
- // TODO
- StatisticsJob pendingJob = pendingJobQueue.peek();
- // step0: check job state again
- // step1: divide statistics job to task
- List<StatisticsTask> statisticsTaskList = divide(pendingJob);
- // step2: submit
-
Catalog.getCurrentCatalog().getStatisticsTaskScheduler().addTasks(statisticsTaskList);
+ StatisticsJob pendingJob = this.pendingJobQueue.peek();
+ if (pendingJob != null) {
+ this.pendingJobQueue.remove();
+ StatisticsJob.JobState jobState = pendingJob.getJobState();
+ // job scheduler is only responsible for moving the job from
pending -> scheduler
+ if (jobState == StatisticsJob.JobState.PENDING) {
+ pendingJob.setScheduleTime(System.currentTimeMillis());
+ try {
+ List<StatisticsTask> tasks = this.divide(pendingJob);
+ ArrayList<StatisticsTask> list = Lists.newArrayList();
+ for (StatisticsTask task : tasks) {
+ // TODO now only support meta task
+ if (task instanceof MetaStatisticsTask){
+ list.add(task);
+ }
+ }
+ pendingJob.setTasks(list);
+
Catalog.getCurrentCatalog().getStatisticsTaskScheduler().addTasks(list);
+ pendingJob.setJobState(StatisticsJob.JobState.SCHEDULING);
+ } catch (DdlException e) {
+ pendingJob.setJobState(StatisticsJob.JobState.FAILED);
+ LOG.info("Failed to schedule the statistical job. " +
e.getMessage());
+ }
+ }
+ }
}
public void addPendingJob(StatisticsJob statisticsJob) throws
IllegalStateException {
- pendingJobQueue.add(statisticsJob);
+ this.pendingJobQueue.add(statisticsJob);
+ }
+
+ /**
+ * Statistics tasks are of the following types：
+ * table:
+ * - row_count: table row count are critical in estimating cardinality and
memory usage of scan nodes.
+ * - data_size: table size, not applicable to CBO, mainly used to monitor
and manage table size.
+ * column:
+ * - num_distinct_value: used to determine the selectivity of an
equivalent expression.
+ * - min: The minimum value.
+ * - max: The maximum value.
+ * - num_nulls: number of nulls.
+ * - avg_col_len: the average length of a column, in bytes, is used for
memory and network IO evaluation.
+ * - max_col_len: the Max length of the column, in bytes, is used for
memory and network IO evaluation.
+ * <p>
+ * Divide:
+ * - min, max, ndv: These three full indicators are collected by a
sub-task.
+ * - max_col_lens, avg_col_lens: Two sampling indicators were collected by
a sub-task.
+ * <p>
+ * If the table row-count is greater than the maximum number of Be scans
for a single BE,
+ * we'll divide subtasks by partition. relevant
values(3700000000L&600000000L) are derived from test.
+ * <p>
+ * Eventually, we will get several subtasks of the following types:
+ *
+ * @throws DdlException DdlException
+ * @see MetaStatisticsTask
+ * @see SampleSQLStatisticsTask
+ * @see SQLStatisticsTask
+ */
+ private List<StatisticsTask> divide(StatisticsJob statisticsJob) throws
DdlException {
+ long jobId = statisticsJob.getId();
+ long dbId = statisticsJob.getDbId();
+ Database db = Catalog.getCurrentCatalog().getDbOrDdlException(dbId);
+ Set<Long> tableIds = statisticsJob.relatedTableId();
+ Map<Long, List<String>> tableIdToColumnName =
statisticsJob.getTableIdToColumnName();
+ List<StatisticsTask> tasks = statisticsJob.getTasks();
+ List<Long> backendIds =
Catalog.getCurrentSystemInfo().getBackendIds(true);
+
+ for (Long tableId : tableIds) {
+ Table tbl = db.getTableOrDdlException(tableId);
+ long rowCount = tbl.getRowCount();
+ List<Long> partitionIds = ((OlapTable) tbl).getPartitionIds();
+ List<String> columnNameList = tableIdToColumnName.get(tableId);
+
+ // step 1: generate data_size task
+ StatsCategoryDesc dataSizeCategory =
this.getTblStatsCategoryDesc(dbId, tableId);
+ StatsGranularityDesc dataSizeGranularity =
this.getTblStatsGranularityDesc(tableId);
+ MetaStatisticsTask dataSizeTask = new MetaStatisticsTask(jobId,
+ dataSizeGranularity, dataSizeCategory,
Collections.singletonList(StatsType.DATA_SIZE));
+ tasks.add(dataSizeTask);
+
+ // step 2: generate row_count task
+ KeysType keysType = ((OlapTable) tbl).getKeysType();
+ if (keysType == KeysType.DUP_KEYS) {
+ StatsCategoryDesc rowCountCategory =
this.getTblStatsCategoryDesc(dbId, tableId);
+ StatsGranularityDesc rowCountGranularity =
this.getTblStatsGranularityDesc(tableId);
+ MetaStatisticsTask metaTask = new MetaStatisticsTask(jobId,
+ rowCountGranularity, rowCountCategory,
Collections.singletonList(StatsType.ROW_COUNT));
+ tasks.add(metaTask);
+ } else {
+ if (rowCount > backendIds.size() * 3700000000L) {
Review Comment:
got it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
