sql: add ImportType to descriptor

This adds an ImportType field to table descriptors. We need this field
if we want to support disabling ImportEpoch writing via a cluster
setting since we won't necessarily be able to trust that a non-zero
ImportEpoch is actually from the currently in-progress IMPORT.

Epic: none
Release note: None

pkg/ccl/backupccl/restore_job.go

@@ -830,7 +830,9 @@ func epochBasedInProgressImport(desc catalog.Descriptor) bool {
 		return false
 	}
 
-	return table.GetInProgressImportStartTime() > 0 && table.TableDesc().ImportEpoch > 0
+	return table.GetInProgressImportStartTime() > 0 &&
+		table.TableDesc().ImportEpoch > 0 &&
+		table.TableDesc().ImportType == descpb.ImportType_IMPORT_WITH_IMPORT_EPOCH
 }
 
 // createImportingDescriptors creates the tables that we will restore into and returns up to three

pkg/ccl/backupccl/testdata/backup-restore/import-epoch

@@ -9,7 +9,6 @@ exec-sql
 SET CLUSTER SETTING bulkio.import.write_import_epoch.enabled=true;
 ----
 
-
 exec-sql
 CREATE DATABASE d;
 USE d;
@@ -19,11 +18,11 @@ INSERT INTO baz VALUES (1, 'x'),(2,'y'),(3,'z');
 ----
 
 exec-sql
-CREATE VIEW import_epoch (epoch)
+CREATE VIEW import_epoch (epoch, type)
 AS WITH tbls AS (
    	SELECT id, crdb_internal.pb_to_json('cockroach.sql.sqlbase.Descriptor', descriptor) AS orig FROM system.descriptor
    )
-   SELECT orig->'table'->'importEpoch' FROM tbls WHERE id = '109';
+   SELECT orig->'table'->'importEpoch', orig->'table'->'importType' FROM tbls WHERE id = '109';
 ----
 
 exec-sql
@@ -43,7 +42,7 @@ foo 109
 query-sql
 SELECT * FROM import_epoch
 ----
-1
+1 <nil>
 
 exec-sql
 EXPORT INTO CSV 'nodelocal://1/export2/' FROM SELECT * FROM baz WHERE i = 2;
@@ -57,7 +56,7 @@ IMPORT INTO foo (i,s) CSV DATA ('nodelocal://1/export2/export*-n*.0.csv')
 query-sql
 SELECT * FROM import_epoch
 ----
-2
+2 <nil>
 
 exec-sql
 SET CLUSTER SETTING jobs.debug.pausepoints = 'import.after_ingest';
@@ -78,13 +77,31 @@ job paused at pausepoint
 query-sql
 SELECT * FROM import_epoch
 ----
-3
+3 "IMPORT_WITH_IMPORT_EPOCH"
 
 # Cancel the job so that the cleanup hook runs.
 job cancel=a
 ----
 
+# Ensure that the import type is cleared.
+query-sql
+SELECT * FROM import_epoch
+----
+3 <nil>
+
+
+exec-sql
+SET CLUSTER SETTING bulkio.import.write_import_epoch.enabled=false;
+----
+
+# Ensure that the correct ImportType is set when the cluster setting is disabled
+import expect-pausepoint tag=b
+IMPORT INTO foo (i,s) CSV DATA ('nodelocal://1/export3/export*-n*.0.csv')
+----
+job paused at pausepoint
+
+# We expect the zero-valued version here.
 query-sql
 SELECT * FROM import_epoch
 ----
-3
+3 <nil>

pkg/sql/catalog/bootstrap/bootstrap_test.go

@@ -154,7 +154,7 @@ func TestSystemDatabaseSchemaBootstrapVersionBumped(t *testing.T) {
 
 	// If you need to update this value (i.e. failed this test), check whether
 	// you need to bump systemschema.SystemDatabaseSchemaBootstrapVersion too.
-	const prevSystemHash = "2c6957d6e966665567ddf47ccc1f465b11b475b4341db308ed0c77524e6f5117"
+	const prevSystemHash = "104f55c10a62f501888678d7fbc987fbae51457307a3e9b23d66a5078e7d48a4"
 	_, curSystemHash := GetAndHashInitialValuesToString(0 /* tenantID */)
 
 	if prevSystemHash != curSystemHash {

pkg/sql/catalog/descpb/structured.proto

@@ -1210,7 +1210,28 @@ message TableDescriptor {
   // gets incremented while preparing the table for ingestion.
   optional uint32 import_epoch = 59 [(gogoproto.nullable) = false, (gogoproto.customname) = "ImportEpoch"];
 
-  // Next ID: 60
+  // ImportType is the type of import that is running if
+  // ImportStartWallTime is set.
+  optional ImportType import_type = 60 [(gogoproto.nullable) = false, (gogoproto.customname) = "ImportType"];
+
+  // Next ID: 61
+}
+
+// ImportType indicates the type of IMPORT that is in progress for a
+// TableDescriptor.
+enum ImportType {
+  // IMPORT_WITH_START_TIME_ONLY indicates that this import set the
+  // ImportStartWallTime on the descriptor but did not set increment
+  // the ImportEpoch. Such imports cannot be rolled back without MVCC
+  // timestamp filtering.
+  //
+  // Note this is purposefuly 0 since in progress imports from before
+  // this field was added won't have this set.
+  IMPORT_WITH_START_TIME_ONLY = 0;
+  // IMPORT_WITH_IMPORT_EPOCH indicates that the running import did
+  // increment the import epoch. Such imports can be rolled back using
+  // an ImportEpoch deletion predicate.
+  IMPORT_WITH_IMPORT_EPOCH = 1;
 }
 
 // SurvivalGoal is the survival goal for a database.

pkg/sql/catalog/tabledesc/table_desc.go

@@ -210,6 +210,7 @@ func (desc *Mutable) SetPublicNonPrimaryIndex(indexOrdinal int, index descpb.Ind
 // import, bumping the ImportEpoch.
 func (desc *Mutable) OfflineForImport() {
 	desc.SetOffline(OfflineReasonImporting)
+	desc.ImportType = descpb.ImportType_IMPORT_WITH_IMPORT_EPOCH
 	desc.ImportEpoch++
 }
 
@@ -224,6 +225,7 @@ func (desc *Mutable) InitializeImport(startWallTime int64) error {
 
 // FinalizeImport removes in progress import metadata from the descriptor
 func (desc *Mutable) FinalizeImport() {
+	desc.ImportType = 0
 	desc.ImportStartWallTime = 0
 }
 

pkg/sql/catalog/tabledesc/validate_test.go

@@ -139,6 +139,7 @@ var validationMap = []struct {
 			"HistogramSamples":              {status: thisFieldReferencesNoObjects},
 			"SchemaLocked":                  {status: thisFieldReferencesNoObjects},
 			"ImportEpoch":                   {status: thisFieldReferencesNoObjects},
+			"ImportType":                    {status: thisFieldReferencesNoObjects},
 		},
 	},
 	{