Merge #153294

153294: IMPORT: trigger async INSPECT job for validation r=ZhouXing19 a=ZhouXing19

Fixes #146327

We found a bug that async flush can cause silent data corruption for IMPORT, and this commit is to trigger a validation step to alert if similar data corruption happens again. To avoid increasing the latency for IMPORT we trigger an INSPECT job that runs the validation asynchronously right after the table is set to public at the end of an import. We use the commit timestamp of the internal txn which update the table state to public, so that we can ensure the INSPECT focus on the snapshot at the end of the IMPORT job.

This behavior is gated by cluster settings `bulkio.import.row_count_validation.unsafe.enabled`, which is default to false.

See more internal discussion [here](https://cockroachlabs.slack.com/archives/C02DSDS9TM1/p1757439225277989?thread_ts=1756991224.633879&cid=C02DSDS9TM1).

Release note: Added `bulkio.import.row_count_validation.unsafe.enabled` (default false) that will trigger asynchronous INSPECT job at the end of an IMPORT execution.

Co-authored-by: ZhouXing19 <[email protected]>

Author: craig[bot]

pkg/sql/importer/import_job.go

@@ -107,6 +107,18 @@ var performConstraintValidation = settings.RegisterBoolSetting(
 	settings.WithUnsafe,
 )
 
+// TODO(janexing): tune the default to True when INSPECT is merged in stable release.
+var importRowCountValidation = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"bulkio.import.row_count_validation.unsafe.enabled",
+	"enables asynchronous validation of imported data via INSPECT "+
+		"jobs. When enabled, an INSPECT job runs after import completion to "+
+		"detect potential data corruption. Disabling this setting may result "+
+		"in undetected data corruption if the import process fails.",
+	false,
+	settings.WithUnsafe,
+)
+
 // Resume is part of the jobs.Resumer interface.
 func (r *importResumer) Resume(ctx context.Context, execCtx interface{}) error {
 	p := execCtx.(sql.JobExecContext)
@@ -279,10 +291,28 @@ func (r *importResumer) Resume(ctx context.Context, execCtx interface{}) error {
 		return err
 	}
 
-	if err := r.publishTables(ctx, p.ExecCfg(), res); err != nil {
+	setPublicTimestamp, err := r.publishTables(ctx, p.ExecCfg(), res)
+	if err != nil {
 		return err
 	}
 
+	if importRowCountValidation.Get(&p.ExecCfg().Settings.SV) {
+		tblDesc := tabledesc.NewBuilder(details.Tables[0].Desc).BuildImmutableTable()
+		if len(tblDesc.PublicNonPrimaryIndexes()) > 0 {
+			_, err := sql.TriggerInspectJob(
+				ctx,
+				fmt.Sprintf("import-validation-%s", tblDesc.GetName()),
+				p.ExecCfg(),
+				tblDesc,
+				setPublicTimestamp,
+			)
+			if err != nil {
+				return errors.Wrapf(err, "failed to trigger inspect for import validation for table %s", tblDesc.GetName())
+			}
+			log.Eventf(ctx, "triggered inspect job for import validation for table %s with AOST %s", tblDesc.GetName(), setPublicTimestamp)
+		}
+	}
+
 	emitImportJobEvent(ctx, p, jobs.StateSucceeded, r.job)
 
 	addToFileFormatTelemetry(details.Format.Format.String(), "succeeded")
@@ -304,7 +334,6 @@ func (r *importResumer) Resume(ctx context.Context, execCtx interface{}) error {
 	}
 
 	logutil.LogJobCompletion(ctx, importJobRecoveryEventType, r.job.ID(), true, nil, r.res.Rows)
-
 	return nil
 }
 
@@ -422,21 +451,24 @@ func bindTableDescImportProperties(
 // publishTables updates the status of imported tables from OFFLINE to PUBLIC.
 func (r *importResumer) publishTables(
 	ctx context.Context, execCfg *sql.ExecutorConfig, res kvpb.BulkOpSummary,
-) error {
+) (hlc.Timestamp, error) {
+	var setPublicTimestamp hlc.Timestamp
 	details := r.job.Details().(jobspb.ImportDetails)
 	// Tables should only be published once.
 	if details.TablesPublished {
-		return nil
+		return setPublicTimestamp, nil
 	}
 
 	log.Event(ctx, "making tables live")
 
+	var kvTxn *kv.Txn
 	err := sql.DescsTxn(ctx, execCfg, func(
 		ctx context.Context, txn isql.Txn, descsCol *descs.Collection,
 	) error {
-		b := txn.KV().NewBatch()
+		kvTxn = txn.KV()
+		b := kvTxn.NewBatch()
 		for _, tbl := range details.Tables {
-			newTableDesc, err := descsCol.MutableByID(txn.KV()).Table(ctx, tbl.Desc.ID)
+			newTableDesc, err := descsCol.MutableByID(kvTxn).Table(ctx, tbl.Desc.ID)
 			if err != nil {
 				return err
 			}
@@ -473,7 +505,7 @@ func (r *importResumer) publishTables(
 				return errors.Wrapf(err, "publishing table %d", newTableDesc.ID)
 			}
 		}
-		if err := txn.KV().Run(ctx, b); err != nil {
+		if err := kvTxn.Run(ctx, b); err != nil {
 			return errors.Wrap(err, "publishing tables")
 		}
 
@@ -486,7 +518,16 @@ func (r *importResumer) publishTables(
 		return nil
 	})
 	if err != nil {
-		return err
+		return setPublicTimestamp, err
+	}
+
+	// Try to get the commit timestamp for the transaction that set
+	// the table to public. This timestamp will be used to run the
+	// INSPECT job for this table, to ensure the INSPECT looks at the
+	// snapshot that IMPORT just finished.
+	setPublicTimestamp, err = kvTxn.CommitTimestamp()
+	if err != nil {
+		return setPublicTimestamp, err
 	}
 
 	// Initiate a run of CREATE STATISTICS. We don't know the actual number of
@@ -497,7 +538,7 @@ func (r *importResumer) publishTables(
 		execCfg.StatsRefresher.NotifyMutation(desc, math.MaxInt32 /* rowsAffected */)
 	}
 
-	return nil
+	return setPublicTimestamp, nil
 }
 
 // checkVirtualConstraints checks constraints that are enforced via runtime

pkg/sql/scrub.go

@@ -477,19 +477,38 @@ func (n *scrubNode) runScrubTableJob(
 	ctx context.Context, p *planner, tableDesc catalog.TableDescriptor, asOf hlc.Timestamp,
 ) error {
 	// Consistency check is done async via a job.
-	jobID := p.ExecCfg().JobRegistry.MakeJobID()
+	jobID, err := TriggerInspectJob(ctx, tree.Serialize(n.n), p.ExecCfg(), tableDesc, asOf)
+	if err != nil {
+		return err
+	}
+	// Let the eval context track this job ID for status and error reporting.
+	p.extendedEvalCtx.jobs.addCreatedJobID(jobID)
+	return nil
+}
+
+// TriggerInspectJob starts an inspect job for the snapshot.
+func TriggerInspectJob(
+	ctx context.Context,
+	jobRecordDescription string,
+	execCfg *ExecutorConfig,
+	tableDesc catalog.TableDescriptor,
+	asOf hlc.Timestamp,
+) (jobspb.JobID, error) {
+	// Consistency check is done async via a job.
+	jobID := execCfg.JobRegistry.MakeJobID()
 
 	// TODO(148300): just grab the first secondary index and use that for the
 	// consistency check.
 	// TODO(148365): When INSPECT is added, we want to skip unsupported indexes
 	// and return a NOTICE.
 	secIndexes := tableDesc.PublicNonPrimaryIndexes()
 	if len(secIndexes) == 0 {
-		return errors.AssertionFailedf("must have at least one secondary index")
+		return jobID, errors.AssertionFailedf("must have at least one secondary index")
 	}
 
+	// TODO(sql-queries): add row count check when that is implemented.
 	jr := jobs.Record{
-		Description: tree.Serialize(n.n),
+		Description: jobRecordDescription,
 		Details: jobspb.InspectDetails{
 			Checks: []*jobspb.InspectDetails_Check{
 				{
@@ -507,22 +526,19 @@ func (n *scrubNode) runScrubTableJob(
 	}
 
 	var sj *jobs.StartableJob
-	if err := p.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) (err error) {
-		return p.ExecCfg().JobRegistry.CreateStartableJobWithTxn(ctx, &sj, jobID, txn, jr)
+	if err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) (err error) {
+		return execCfg.JobRegistry.CreateStartableJobWithTxn(ctx, &sj, jobID, txn, jr)
 	}); err != nil {
 		if sj != nil {
 			if cleanupErr := sj.CleanupOnRollback(ctx); cleanupErr != nil {
 				log.Dev.Warningf(ctx, "failed to cleanup StartableJob: %v", cleanupErr)
 			}
 		}
-		return err
+		return jobID, err
 	}
-
-	log.Dev.Infof(ctx, "created and started inspect job %d (no-op)", jobID)
+	log.Dev.Infof(ctx, "created and started inspect job %d", jobID)
 	if err := sj.Start(ctx); err != nil {
-		return err
+		return jobID, err
 	}
-	// Let the eval context track this job ID for status and error reporting.
-	p.extendedEvalCtx.jobs.addCreatedJobID(jobID)
-	return nil
+	return jobID, nil
 }