infoschema/executor : add DDLJobs sys table

executor/builder.go

@@ -1365,6 +1365,11 @@ func (b *executorBuilder) buildMemTable(v *plannercore.PhysicalMemTable) Executo
 				baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID()),
 				retriever:    retriever,
 			}
+		case strings.ToLower(infoschema.TableDDLJobs):
+			return &DDLJobsReaderExec{
+				baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ExplainID()),
+				is:           b.is,
+			}
 		}
 	}
 	tb, _ := b.is.TableByID(v.Table.ID)

executor/executor.go

@@ -339,14 +339,34 @@ func (e *ShowDDLExec) Next(ctx context.Context, req *chunk.Chunk) error {
 // ShowDDLJobsExec represent a show DDL jobs executor.
 type ShowDDLJobsExec struct {
 	baseExecutor
+	DDLJobExecInitializer
 
-	cursor         int
+	cacheJobs []*model.Job
+	jobNumber int
+	is        infoschema.InfoSchema
+	done      bool
+}
+
+// DDLJobExecInitializer initials the DDLJobsExec
+type DDLJobExecInitializer struct {
 	runningJobs    []*model.Job
 	historyJobIter *meta.LastJobIterator
-	cacheJobs      []*model.Job
-	jobNumber      int
-	is             infoschema.InfoSchema
-	done           bool
+	cursor         int
+}
+
+func (e *DDLJobExecInitializer) initial(txn kv.Transaction) error {
+	jobs, err := admin.GetDDLJobs(txn)
+	if err != nil {
+		return err
+	}
+	m := meta.NewMeta(txn)
+	e.historyJobIter, err = m.GetLastHistoryDDLJobsIterator()
+	if err != nil {
+		return err
+	}
+	e.runningJobs = jobs
+	e.cursor = 0
+	return nil
 }
 
 // ShowDDLJobQueriesExec represents a show DDL job queries executor.
@@ -414,21 +434,13 @@ func (e *ShowDDLJobsExec) Open(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	jobs, err := admin.GetDDLJobs(txn)
-	if err != nil {
-		return err
-	}
 	if e.jobNumber == 0 {
 		e.jobNumber = admin.DefNumHistoryJobs
 	}
-
-	m := meta.NewMeta(txn)
-	e.historyJobIter, err = m.GetLastHistoryDDLJobsIterator()
+	err = e.DDLJobExecInitializer.initial(txn)
 	if err != nil {
 		return err
 	}
-	e.runningJobs = append(e.runningJobs, jobs...)
-	e.cursor = 0
 	return nil
 }
 
@@ -439,6 +451,7 @@ func (e *ShowDDLJobsExec) Next(ctx context.Context, req *chunk.Chunk) error {
 		return nil
 	}
 	count := 0
+
 	// Append running ddl jobs.
 	if e.cursor < len(e.runningJobs) {
 		numCurBatch := mathutil.Min(req.Capacity(), len(e.runningJobs)-e.cursor)
@@ -448,13 +461,14 @@ func (e *ShowDDLJobsExec) Next(ctx context.Context, req *chunk.Chunk) error {
 		e.cursor += numCurBatch
 		count += numCurBatch
 	}
-	var err error
+
 	// Append history ddl jobs.
+	var err error
 	if count < req.Capacity() {
 		num := req.Capacity() - count
 		remainNum := e.jobNumber - (e.cursor - len(e.runningJobs))
 		num = mathutil.Min(num, remainNum)
-		e.cacheJobs, err = e.historyJobIter.GetLastJobs(num, e.cacheJobs)
+		e.cacheJobs, err = e.historyJobIter.GetNextJobs(num, e.cacheJobs)
 		if err != nil {
 			return err
 		}

executor/infoschema_reader.go

@@ -15,14 +15,16 @@ package executor
 
 import (
 	"context"
-	"sort"
-
+	"github.com/cznic/mathutil"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/privilege"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/chunk"
+	"sort"
+	"strings"
 )
 
 type memtableRetriever struct {
@@ -153,3 +155,103 @@ func dataForViews(ctx sessionctx.Context, schemas []*model.DBInfo) ([][]types.Da
 	}
 	return rows, nil
 }
+
+// DDLJobsReaderExec executes DDLJobs information retrieving
+type DDLJobsReaderExec struct {
+	baseExecutor
+	DDLJobExecInitializer
+
+	cacheJobs []*model.Job
+	is        infoschema.InfoSchema
+}
+
+// Open implements the Executor Next interface.
+func (e *DDLJobsReaderExec) Open(ctx context.Context) error {
+	if err := e.baseExecutor.Open(ctx); err != nil {
+		return err
+	}
+	txn, err := e.ctx.Txn(true)
+	if err != nil {
+		return err
+	}
+	err = e.DDLJobExecInitializer.initial(txn)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Next implements the Executor Next interface.
+func (e *DDLJobsReaderExec) Next(ctx context.Context, req *chunk.Chunk) error {
+	req.GrowAndReset(e.maxChunkSize)
+	checker := privilege.GetPrivilegeManager(e.ctx)
+	count := 0
+
+	// Append running ddl jobs.
+	if e.cursor < len(e.runningJobs) {
+		num := mathutil.Min(req.Capacity(), len(e.runningJobs)-e.cursor)
+		for i := e.cursor; i < e.cursor+num; i++ {
+			e.appendJobToChunk(req, e.runningJobs[i], checker)
+		}
+		e.cursor += num
+		count += num
+	}
+	var err error
+
+	// Append history ddl jobs.
+	if count < req.Capacity() {
+		e.cacheJobs, err = e.historyJobIter.GetNextJobs(req.Capacity()-count, e.cacheJobs)
+		if err != nil {
+			return err
+		}
+		for _, job := range e.cacheJobs {
+			e.appendJobToChunk(req, job, checker)
+		}
+		e.cursor += len(e.cacheJobs)
+	}
+	return nil
+}
+
+func (e *DDLJobsReaderExec) appendJobToChunk(req *chunk.Chunk, job *model.Job, checker privilege.Manager) {
+	schemaName := job.SchemaName
+	tableName := ""
+	finishTS := uint64(0)
+	if job.BinlogInfo != nil {
+		finishTS = job.BinlogInfo.FinishedTS
+		if job.BinlogInfo.TableInfo != nil {
+			tableName = job.BinlogInfo.TableInfo.Name.L
+		}
+		if len(schemaName) == 0 && job.BinlogInfo.DBInfo != nil {
+			schemaName = job.BinlogInfo.DBInfo.Name.L
+		}
+	}
+	// For compatibility, the old version of DDL Job wasn't store the schema name and table name.
+	if len(schemaName) == 0 {
+		schemaName = getSchemaName(e.is, job.SchemaID)
+	}
+	if len(tableName) == 0 {
+		tableName = getTableName(e.is, job.TableID)
+	}
+
+	//check the privilege
+	if checker != nil && !checker.RequestVerification(e.ctx.GetSessionVars().ActiveRoles, strings.ToLower(schemaName), strings.ToLower(tableName), "", mysql.AllPrivMask) {
+		return
+	}
+
+	req.AppendInt64(0, job.ID)
+	req.AppendString(1, schemaName)
+	req.AppendString(2, tableName)
+	req.AppendString(3, job.Type.String())
+	req.AppendString(4, job.SchemaState.String())
+	req.AppendInt64(5, job.SchemaID)
+	req.AppendInt64(6, job.TableID)
+	req.AppendInt64(7, job.RowCount)
+	req.AppendString(8, model.TSConvert2Time(job.StartTS).String())
+	if finishTS > 0 {
+		req.AppendString(9, model.TSConvert2Time(finishTS).String())
+	} else {
+		req.AppendString(9, "")
+	}
+	req.AppendString(10, job.State.String())
+	req.AppendString(11, job.Query)
+}

executor/infoschema_reader_test.go

@@ -83,3 +83,36 @@ func (s *testInfoschemaTableSuite) TestViews(c *C) {
 	tk.MustQuery("SELECT * FROM information_schema.views WHERE table_schema='test' AND table_name='v1'").Check(testkit.Rows("def test v1 SELECT 1 CASCADED NO root@localhost DEFINER utf8mb4 utf8mb4_bin"))
 	tk.MustQuery("SELECT table_catalog, table_schema, table_name, table_type, engine, version, row_format, table_rows, avg_row_length, data_length, max_data_length, index_length, data_free, auto_increment, update_time, check_time, table_collation, checksum, create_options, table_comment FROM information_schema.tables WHERE table_schema='test' AND table_name='v1'").Check(testkit.Rows("def test v1 VIEW <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> VIEW"))
 }
+
+func (s *testInfoschemaTableSuite) TestDDLJobs(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("create database if not exists test_ddl_jobs")
+	tk.MustQuery("select db_name, job_type from information_schema.DDL_JOBS limit 1").Check(
+		testkit.Rows("test_ddl_jobs create schema"))
+
+	tk.MustExec("use test_ddl_jobs")
+	tk.MustExec("create table t (a int);")
+	tk.MustQuery("select db_name, table_name, job_type from information_schema.DDL_JOBS where table_name = 't'").Check(
+		testkit.Rows("test_ddl_jobs t create table"))
+
+	tk.MustQuery("select job_type from information_schema.DDL_JOBS group by job_type having job_type = 'create table'").Check(
+		testkit.Rows("create table"))
+
+	//test the privilege of new user for information_schema.DDL_JOBS
+	tk.MustExec("create user DDL_JOBS_tester")
+	DDLJobsTester := testkit.NewTestKit(c, s.store)
+	DDLJobsTester.MustExec("use information_schema")
+	c.Assert(DDLJobsTester.Se.Auth(&auth.UserIdentity{
+		Username: "DDL_JOBS_tester",
+		Hostname: "127.0.0.1",
+	}, nil, nil), IsTrue)
+	DDLJobsTester.MustQuery("select DB_NAME from information_schema.DDL_JOBS;").Check(
+		[][]interface{}{})
+
+	//test the privilege of user with privilege of mysql for information_schema.ddl_jobs
+	tk.MustExec("CREATE ROLE mysql_priv;")
+	tk.MustExec("GRANT ALL PRIVILEGES ON mysql.* TO mysql_priv;")
+	tk.MustExec("GRANT mysql_priv TO DDL_JOBS_tester;")
+	DDLJobsTester.MustExec("set role mysql_priv")
+	DDLJobsTester.MustQuery("select DISTINCT DB_NAME from information_schema.DDL_JOBS where DB_NAME = 'mysql';").Check(testkit.Rows("mysql"))
+}

infoschema/tables.go

@@ -119,6 +119,8 @@ const (
 	TableMetricSummaryByLabel = "METRICS_SUMMARY_BY_LABEL"
 	// TableInspectionSummary is the string constant of inspection summary table
 	TableInspectionSummary = "INSPECTION_SUMMARY"
+	// TableDDLJobs is the string constant of DDL job table.
+	TableDDLJobs = "DDL_JOBS"
 )
 
 var tableIDMap = map[string]int64{
@@ -177,6 +179,7 @@ var tableIDMap = map[string]int64{
 	TableMetricSummaryByLabel:               autoid.InformationSchemaDBID + 53,
 	TableMetricTables:                       autoid.InformationSchemaDBID + 54,
 	TableInspectionSummary:                  autoid.InformationSchemaDBID + 55,
+	TableDDLJobs:                            autoid.InformationSchemaDBID + 56,
 }
 
 type columnInfo struct {
@@ -841,6 +844,21 @@ var tableClusterSystemInfoCols = []columnInfo{
 	{"VALUE", mysql.TypeVarchar, 128, 0, nil, nil},
 }
 
+var tableDDLJobsCols = []columnInfo{
+	{"JOB_ID", mysql.TypeLonglong, 4, 0, nil, nil},
+	{"DB_NAME", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"TABLE_NAME", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"JOB_TYPE", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"SCHEMA_STATE", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"SCHEMA_ID", mysql.TypeLonglong, 4, 0, nil, nil},
+	{"TABLE_ID", mysql.TypeLonglong, 4, 0, nil, nil},
+	{"ROW_COUNT", mysql.TypeLonglong, 4, 0, nil, nil},
+	{"START_TIME", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"END_TIME", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"STATE", mysql.TypeVarchar, 64, 0, nil, nil},
+	{"QUERY", mysql.TypeVarchar, 64, 0, nil, nil},
+}
+
 func dataForTiKVRegionStatus(ctx sessionctx.Context) (records [][]types.Datum, err error) {
 	tikvStore, ok := ctx.GetStore().(tikv.Storage)
 	if !ok {
@@ -2449,6 +2467,7 @@ var tableNameToColumns = map[string][]columnInfo{
 	TableMetricSummaryByLabel:               tableMetricSummaryByLabelCols,
 	TableMetricTables:                       tableMetricTablesCols,
 	TableInspectionSummary:                  tableInspectionSummaryCols,
+	TableDDLJobs:                            tableDDLJobsCols,
 }
 
 func createInfoSchemaTable(_ autoid.Allocators, meta *model.TableInfo) (table.Table, error) {

meta/meta.go

@@ -779,8 +779,8 @@ func (m *Meta) GetLastHistoryDDLJobsIterator() (*LastJobIterator, error) {
 	}, nil
 }
 
-// GetLastJobs gets last several jobs.
-func (i *LastJobIterator) GetLastJobs(num int, jobs []*model.Job) ([]*model.Job, error) {
+// GetNextJobs gets next several jobs.
+func (i *LastJobIterator) GetNextJobs(num int, jobs []*model.Job) ([]*model.Job, error) {
 	if len(jobs) < num {
 		jobs = make([]*model.Job, 0, num)
 	}