Merge pull request #7781 from vivekmenezes/vivek/extend

sql: ExtendLease() returns existing valid unexpired lease

sql/schema_changer.go

@@ -35,6 +35,15 @@ import (
 	"github.com/pkg/errors"
 )
 
+var (
+	// SchemaChangeLeaseDuration is the duration a lease will be acquired for.
+	// Exported for testing purposes only.
+	SchemaChangeLeaseDuration = 5 * time.Minute
+	// MinSchemaChangeLeaseDuration is the minimum duration a lease will have
+	// remaining upon acquisition. Exported for testing purposes only.
+	MinSchemaChangeLeaseDuration = time.Minute
+)
+
 // SchemaChanger is used to change the schema on a table.
 type SchemaChanger struct {
 	tableID    sqlbase.ID
@@ -78,7 +87,7 @@ func NewSchemaChangerForTesting(
 
 func (sc *SchemaChanger) createSchemaChangeLease() sqlbase.TableDescriptor_SchemaChangeLease {
 	return sqlbase.TableDescriptor_SchemaChangeLease{
-		NodeID: sc.nodeID, ExpirationTime: timeutil.Now().Add(jitteredLeaseDuration()).UnixNano()}
+		NodeID: sc.nodeID, ExpirationTime: timeutil.Now().Add(SchemaChangeLeaseDuration).UnixNano()}
 }
 
 var errExistingSchemaChangeLease = errors.New(
@@ -146,16 +155,25 @@ func (sc *SchemaChanger) ReleaseLease(lease sqlbase.TableDescriptor_SchemaChange
 	return err
 }
 
-// ExtendLease for the current leaser.
+// ExtendLease for the current leaser. This needs to be called often while
+// doing a schema change to prevent more than one node attempting to apply a
+// schema change (which is still safe, but unwise).
 func (sc *SchemaChanger) ExtendLease(
 	existingLease sqlbase.TableDescriptor_SchemaChangeLease,
 ) (sqlbase.TableDescriptor_SchemaChangeLease, error) {
+	// Check if there is still time on this lease.
+	minDesiredExpiration := timeutil.Now().Add(MinSchemaChangeLeaseDuration)
+	if time.Unix(0, existingLease.ExpirationTime).After(minDesiredExpiration) {
+		return existingLease, nil
+	}
+	// Update lease.
 	var lease sqlbase.TableDescriptor_SchemaChangeLease
 	err := sc.db.Txn(func(txn *client.Txn) error {
 		tableDesc, err := sc.findTableWithLease(txn, existingLease)
 		if err != nil {
 			return err
 		}
+
 		lease = sc.createSchemaChangeLease()
 		tableDesc.Lease = &lease
 		txn.SetSystemConfigTrigger()

sql/schema_changer_test.go

@@ -54,6 +54,12 @@ func TestSchemaChangeLease(t *testing.T) {
 	params, _ := createTestServerParams()
 	s, sqlDB, kvDB := serverutils.StartServer(t, params)
 	defer s.Stopper().Stop()
+	// Set MinSchemaChangeLeaseDuration to always expire the lease.
+	minLeaseDuration := csql.MinSchemaChangeLeaseDuration
+	csql.MinSchemaChangeLeaseDuration = 2 * csql.SchemaChangeLeaseDuration
+	defer func() {
+		csql.MinSchemaChangeLeaseDuration = minLeaseDuration
+	}()
 
 	if _, err := sqlDB.Exec(`
 CREATE DATABASE t;
@@ -78,8 +84,10 @@ CREATE TABLE t.test (k CHAR PRIMARY KEY, v CHAR);
 	}
 
 	// Acquiring another lease will fail.
-	if newLease, err := changer.AcquireLease(); err == nil {
-		t.Fatalf("acquired new lease: %v, while unexpired lease exists: %v", newLease, lease)
+	if _, err := changer.AcquireLease(); !testutils.IsError(
+		err, "an outstanding schema change lease exists",
+	) {
+		t.Fatal(err)
 	}
 
 	// Extend the lease.
@@ -92,10 +100,14 @@ CREATE TABLE t.test (k CHAR PRIMARY KEY, v CHAR);
 		t.Fatalf("invalid expiration time: %s", time.Unix(0, newLease.ExpirationTime))
 	}
 
+	// The new lease is a brand new lease.
+	if newLease == lease {
+		t.Fatalf("lease was not extended: %v", lease)
+	}
+
 	// Extending an old lease fails.
-	_, err = changer.ExtendLease(lease)
-	if err == nil {
-		t.Fatal("extending an old lease succeeded")
+	if _, err := changer.ExtendLease(lease); !testutils.IsError(err, "table: .* has lease") {
+		t.Fatal(err)
 	}
 
 	// Releasing an old lease fails.
@@ -121,6 +133,17 @@ CREATE TABLE t.test (k CHAR PRIMARY KEY, v CHAR);
 	if err != nil {
 		t.Fatal(err)
 	}
+
+	// Set MinSchemaChangeLeaseDuration to not expire the lease.
+	csql.MinSchemaChangeLeaseDuration = minLeaseDuration
+	newLease, err = changer.ExtendLease(lease)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The old lease is renewed.
+	if newLease != lease {
+		t.Fatalf("acquired new lease: %v, old lease: %v", newLease, lease)
+	}
 }
 
 func validExpirationTime(expirationTime int64) bool {