diff --git a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
index 3677e8373d8..b4f1b89617f 100644
--- a/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
+++ b/access-control-service/src/test/scala/org/apache/texera/AccessControlResourceSpec.scala
@@ -66,7 +66,6 @@ class AccessControlResourceSpec
     user.setName("testuser")
     user.setEmail("test@example.com")
     user.setRole(UserRoleEnum.REGULAR)
-    user.setPassword("password")
     user
   }
 
@@ -76,7 +75,6 @@ class AccessControlResourceSpec
     user.setName("testuser2")
     user.setEmail("test2@example.com")
     user.setRole(UserRoleEnum.REGULAR)
-    user.setPassword("password")
     user
   }
 
@@ -139,7 +137,7 @@ class AccessControlResourceSpec
       computingUnitOfUserDao.insert(cuAccess)
     }
 
-    val claims = JwtAuth.jwtClaims(testUser1, 1)
+    val claims = JwtAuth.jwtClaims(testUser1)
     token = JwtAuth.jwtToken(claims)
   }
 
diff --git a/access-control-service/src/test/scala/org/apache/texera/service/resource/LiteLLMProxyAuthSpec.scala b/access-control-service/src/test/scala/org/apache/texera/service/resource/LiteLLMProxyAuthSpec.scala
index 4d8d271c7d6..2fc5c10062a 100644
--- a/access-control-service/src/test/scala/org/apache/texera/service/resource/LiteLLMProxyAuthSpec.scala
+++ b/access-control-service/src/test/scala/org/apache/texera/service/resource/LiteLLMProxyAuthSpec.scala
@@ -164,9 +164,8 @@ class LiteLLMProxyAuthSpec extends AnyFlatSpec with Matchers with BeforeAndAfter
     u.setUid(1)
     u.setName("test")
     u.setEmail("test@example.com")
-    u.setGoogleId(null)
     u.setRole(role)
-    JwtAuth.jwtToken(JwtAuth.jwtClaims(u, expireInDays = 1))
+    JwtAuth.jwtToken(JwtAuth.jwtClaims(u))
   }
 
   private val chatBody = """{"model":"gpt-4o-mini","messages":[]}"""
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/auth/AuthResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/auth/AuthResource.scala
index db443aae8e2..b05ed180bd9 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/auth/AuthResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/auth/AuthResource.scala
@@ -19,19 +19,19 @@
 
 package org.apache.texera.web.resource.auth
 
-import org.apache.texera.auth.JwtAuth.{TOKEN_EXPIRE_TIME_IN_MINUTES, jwtClaims, jwtToken}
+import com.typesafe.scalalogging.Logger
+import org.apache.texera.auth.JwtAuth.{jwtClaims, jwtToken}
 import org.apache.texera.common.config.UserSystemConfig
 import org.apache.texera.common.util.EmailUtil
 import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.Tables.USER
-import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
-import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
 import org.apache.texera.dao.jooq.generated.tables.pojos.User
 import org.apache.texera.web.model.http.request.auth.{UserLoginRequest, UserRegistrationRequest}
 import org.apache.texera.web.model.http.response.TokenIssueResponse
 import org.apache.texera.web.resource.auth.AuthResource._
+import org.jooq.DSLContext
 import org.jooq.impl.DSL
-import org.jasypt.util.password.StrongPasswordEncryptor
 
 import java.time.Instant
 import java.time.temporal.ChronoUnit
@@ -39,53 +39,68 @@ import javax.ws.rs._
 import javax.ws.rs.core.MediaType
 
 object AuthResource {
+  private val logger: Logger = Logger(classOf[AuthResource])
 
-  private def userDao =
-    new UserDao(
-      SqlServer
-        .getInstance()
-        .createDSLContext()
-        .configuration
-    )
+  private def context = SqlServer.getInstance().context
 
   /**
     * Retrieve exactly one User from databases with the given username and password.
     * The password is used to validate against the hashed password stored in the db.
     *
-    * @param name     String
+    * @param username the LOCAL login handle to authenticate
     * @param password String, plain text password
     * @return
     */
-  def retrieveUserByUsernameAndPassword(name: String, password: String): Option[User] = {
-    if (password == null) return None
-    if (name == null) return None
-    Option(
-      SqlServer
-        .getInstance()
-        .createDSLContext()
-        .select()
-        .from(USER)
-        .where(USER.NAME.eq(name))
-        .fetchOneInto(classOf[User])
-    ).filter(user => new StrongPasswordEncryptor().checkPassword(password, user.getPassword))
+  def retrieveUserByUsernameAndPassword(username: String, password: String): Option[User] = {
+    if (password == null || username == null) return None
+
+    val record = context
+      .select()
+      .from(AUTH_PROVIDER)
+      .join(USER)
+      .on(USER.UID.eq(AUTH_PROVIDER.UID))
+      .where(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+      .and(AUTH_PROVIDER.PROVIDER_ID.eq(username))
+      .fetchOne()
+
+    Option(record).flatMap(r => {
+      val encryptedPassword = r.get(AUTH_PROVIDER.PASSWORD)
+      if (LocalAuthProvisioner.checkPassword(password, encryptedPassword)) {
+        Some(r.into(USER).into(classOf[User]))
+      } else {
+        None
+      }
+    })
   }
 
-  /**
-    * Marks a placeholder account (auto-created for a dataset contributor) as
-    * claimed, leaving persistence to the caller.
-    */
   /**
     * Email identity is matched case-insensitively (backed by idx_user_email_lower),
     * while stored emails keep their original casing.
+    *
+    * Case-insensitivity is required, not a nicety: `"user".email` is a plain case-sensitive
+    * UNIQUE and `idx_user_email_lower` is not unique, so `Alice@x.com` and `alice@x.com` can
+    * coexist. Registration stores the address as the user typed it while contributor
+    * placeholders are stored lower-cased, so the casings provably differ in practice. An
+    * exact-match lookup would miss, insert a second account without violating any constraint,
+    * and silently strand the original account's data.
     */
   def fetchUserByEmailIgnoreCase(email: String): User =
-    SqlServer
-      .getInstance()
-      .createDSLContext()
+    fetchUserByEmailIgnoreCase(SqlServer.getInstance().createDSLContext(), email)
+
+  /**
+    * As above, against a caller-supplied context. [[ExternalAuthProvisioner]] passes its
+    * transaction's context so the lookup reads that transaction's own writes.
+    */
+  def fetchUserByEmailIgnoreCase(ctx: DSLContext, email: String): User =
+    ctx
       .selectFrom(USER)
       .where(DSL.lower(USER.EMAIL).eq(EmailUtil.normalize(email)))
       .fetchOneInto(classOf[User])
 
+  /**
+    * Marks a placeholder account (auto-created for a dataset contributor) as
+    * claimed, leaving persistence to the caller.
+    */
   def claimPlaceholder(user: User): Unit = {
     user.setIsPlaceholder(false)
     val claimedAt = Instant.now().truncatedTo(ChronoUnit.SECONDS)
@@ -95,21 +110,33 @@ object AuthResource {
     )
   }
 
-  def createAdminUser(): Unit = {
-    val adminUsername = UserSystemConfig.adminUsername
-    val adminPassword = UserSystemConfig.adminPassword
+  def createAdminUser(): Unit =
+    createAdminUser(UserSystemConfig.adminUsername.trim, UserSystemConfig.adminPassword.trim)
 
-    if (adminUsername.trim.nonEmpty && adminPassword.trim.nonEmpty) {
-      val existingUser = userDao.fetchByName(adminUsername)
-      if (existingUser.isEmpty) {
-        val user = new User
-        user.setName(adminUsername)
-        user.setEmail(adminUsername)
-        user.setRole(UserRoleEnum.ADMIN)
-        user.setPassword(new StrongPasswordEncryptor().encryptPassword(adminPassword))
-        userDao.insert(user)
-      }
+  /**
+    * Bootstrap the configured admin account, doing nothing if it already exists. The credentials
+    * are parameters rather than reads of [[UserSystemConfig]] because those are object vals
+    * resolved once per JVM, which leaves the unconfigured case unreachable from a test.
+    */
+  private[auth] def createAdminUser(adminUsername: String, adminPassword: String): Unit = {
+    if (adminUsername.isEmpty || adminPassword.isEmpty) return
+
+    if (LocalAuthProvisioner.handleExists(adminUsername)) return
+
+    if (fetchUserByEmailIgnoreCase(adminUsername) != null) {
+      logger.warn(
+        s"Not creating the admin account: '$adminUsername' is already used as an email address " +
+          "by an account with no local credential. Grant that account the ADMIN role instead."
+      )
+      return
     }
+
+    val user = new User
+    user.setName(adminUsername)
+    user.setEmail(adminUsername)
+    user.setRole(UserRoleEnum.ADMIN)
+
+    LocalAuthProvisioner.createLocalAccount(user, adminUsername, adminPassword)
   }
 }
 
@@ -123,7 +150,11 @@ class AuthResource {
   def login(request: UserLoginRequest): TokenIssueResponse = {
     retrieveUserByUsernameAndPassword(request.username, request.password) match {
       case Some(user) =>
-        TokenIssueResponse(jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES)))
+        // An account can hold both a LOCAL and a GOOGLE credential, and the frontend expects
+        // `googleId` in the token regardless of which one was used to sign in.
+        val googleId =
+          ExternalAuthProvisioner.providerIdOf(user.getUid, ProviderTypeEnum.GOOGLE)
+        TokenIssueResponse(jwtToken(jwtClaims(user, googleId)))
       case None => throw new NotAuthorizedException("Login credentials are incorrect.")
     }
   }
@@ -143,7 +174,11 @@ class AuthResource {
     if (userpassword == null || userpassword.isEmpty)
       throw new NotAcceptableException("Password cannot be empty")
 
-    val usernameExists = !userDao.fetchByName(username).isEmpty
+    // The username being registered becomes a LOCAL login handle, so the handle is what has to
+    // be free, not the display name. Asking `"user".name` instead both missed genuinely taken
+    // handles (letting the insert die on uq_provider_identity as a 500) and rejected free ones,
+    // because an external login rewrites the display name but never the handle.
+    val usernameExists = LocalAuthProvisioner.handleExists(username)
     val existingByEmail = fetchUserByEmailIgnoreCase(useremail)
     val emailExists = existingByEmail != null
 
@@ -151,12 +186,15 @@ class AuthResource {
     // credential) is claimed by the first registration with its email. The
     // account keeps its uid, so existing contributor links stay valid, and it
     // stays INACTIVE until an admin approves it.
+    //
+    // The credential is written to auth_provider rather than onto the user row, in the same
+    // transaction as the claim, so the account cannot end up marked claimed with nothing to
+    // log in with.
     if (!usernameExists && emailExists && existingByEmail.getIsPlaceholder) {
       existingByEmail.setName(username)
-      existingByEmail.setPassword(new StrongPasswordEncryptor().encryptPassword(userpassword))
       claimPlaceholder(existingByEmail)
-      userDao.update(existingByEmail)
-      return TokenIssueResponse(jwtToken(jwtClaims(existingByEmail, TOKEN_EXPIRE_TIME_IN_MINUTES)))
+      LocalAuthProvisioner.claimWithLocalCredential(existingByEmail, username, userpassword)
+      return TokenIssueResponse(jwtToken(jwtClaims(existingByEmail)))
     }
 
     (usernameExists, emailExists) match {
@@ -169,10 +207,9 @@ class AuthResource {
         user.setName(username)
         user.setEmail(useremail)
         user.setRole(UserRoleEnum.INACTIVE)
-        // hash the plain text password
-        user.setPassword(new StrongPasswordEncryptor().encryptPassword(userpassword))
-        userDao.insert(user)
-        TokenIssueResponse(jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES)))
+        // Reports losing the race to a concurrent registration of the same handle as a 409.
+        LocalAuthProvisioner.createLocalAccount(user, username, userpassword)
+        TokenIssueResponse(jwtToken(jwtClaims(user)))
     }
   }
 
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisioner.scala b/amber/src/main/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisioner.scala
new file mode 100644
index 00000000000..c89d5b72ba9
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisioner.scala
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.auth
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.dao.{SqlServer, SqlStates}
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
+import org.jooq.DSLContext
+import org.jooq.exception.DataAccessException
+
+import java.time.OffsetDateTime
+import scala.util.chaining.scalaUtilChainingOps
+
+/**
+  * A verified external identity (Google, Facebook, ...) reduced to the fields we persist.
+  *
+  * `email` must be non-blank and provider-verified: `loginOrProvision` links the identity to the
+  * account owning that address and claims its placeholder, so an unverified address is a
+  * takeover. Each provider checks this in its own mapping function (Google: `email_verified`).
+  */
+final case class ExternalProfile(
+    providerType: ProviderTypeEnum,
+    providerId: String,
+    name: String,
+    email: String,
+    avatar: String
+)
+
+object ExternalAuthProvisioner extends LazyLogging {
+
+  /**
+    * The account owning `email`, matched case-insensitively and within the caller's transaction
+    * so it reads that transaction's own writes. See
+    * [[AuthResource.fetchUserByEmailIgnoreCase]] for why the match cannot be exact.
+    */
+  private def userByEmailIgnoreCase(ctx: DSLContext, email: String): Option[User] =
+    Option(AuthResource.fetchUserByEmailIgnoreCase(ctx, email))
+
+  /**
+    * Resolve the user behind an external identity, creating one if necessary, and
+    * ensure its auth-provider row is present and up to date. Each attempt runs in one
+    * transaction. A unique violation is taken to mean a concurrent login won the race, so the
+    * attempt is re-run once; if the retry violates a constraint too, that exception propagates.
+    */
+  def loginOrProvision(profile: ExternalProfile): User = {
+    try {
+      provision(profile)
+    } catch {
+      case e: DataAccessException if e.sqlState() == SqlStates.UNIQUE_VIOLATION =>
+        provision(profile)
+    }
+  }
+
+  private def provision(profile: ExternalProfile): User = {
+    SqlServer.withTransaction(SqlServer.getInstance().createDSLContext()) { ctx =>
+      val txUserDao = new UserDao(ctx.configuration())
+      val txAuthDao = new AuthProviderDao(ctx.configuration())
+
+      Option(
+        ctx
+          .select()
+          .from(USER)
+          .join(AUTH_PROVIDER)
+          .on(USER.UID.eq(AUTH_PROVIDER.UID))
+          .where(AUTH_PROVIDER.PROVIDER_TYPE.eq(profile.providerType))
+          .and(AUTH_PROVIDER.PROVIDER_ID.eq(profile.providerId))
+          .fetchOne()
+      ) match {
+        case Some(record) =>
+          // The join above already selected every USER column, so map in place rather than
+          // re-reading the same row by uid.
+          record.into(USER).into(classOf[User]).tap { user =>
+            if (refresh(user, profile)) txUserDao.update(user)
+          }
+
+        case None =>
+          val user = userByEmailIgnoreCase(ctx, profile.email) match {
+            case Some(existing) =>
+              existing.tap { user =>
+                val wasPlaceholder = user.getIsPlaceholder
+                if (wasPlaceholder) AuthResource.claimPlaceholder(user)
+                val drifted = refresh(user, profile)
+                if (drifted || wasPlaceholder) txUserDao.update(user)
+              }
+            case None =>
+              val created = new User()
+              created.setName(profile.name)
+              created.setEmail(profile.email)
+              created.setAvatar(profile.avatar)
+              created.setRole(UserRoleEnum.INACTIVE)
+              txUserDao.insert(created)
+              created
+          }
+          upsertProvider(ctx, txAuthDao, user, profile)
+          user
+      }
+    }
+  }
+
+  /** The external id `uid` authenticates with at `providerType`, if it has one. */
+  def providerIdOf(uid: Integer, providerType: ProviderTypeEnum): Option[String] =
+    Option(
+      SqlServer
+        .getInstance()
+        .context
+        .select(AUTH_PROVIDER.PROVIDER_ID)
+        .from(AUTH_PROVIDER)
+        .where(AUTH_PROVIDER.UID.eq(uid))
+        .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(providerType))
+        .fetchOne(AUTH_PROVIDER.PROVIDER_ID)
+    )
+
+  /**
+    * Mutate `user` in place to match `profile`, returning true iff anything changed
+    * (so the caller only issues an UPDATE when needed).
+    */
+  private def refresh(user: User, profile: ExternalProfile): Boolean = {
+    var changed = false
+    if (user.getName != profile.name) {
+      user.setName(profile.name)
+      changed = true
+    }
+    if (user.getEmail != profile.email) {
+      user.setEmail(profile.email)
+      changed = true
+    }
+    if (user.getAvatar != profile.avatar) {
+      user.setAvatar(profile.avatar)
+      changed = true
+    }
+    changed
+  }
+
+  private def upsertProvider(
+      ctx: DSLContext,
+      authDao: AuthProviderDao,
+      user: User,
+      profile: ExternalProfile
+  ): Unit = {
+    val hasProvider = ctx.fetchExists(
+      ctx
+        .selectFrom(AUTH_PROVIDER)
+        .where(AUTH_PROVIDER.UID.eq(user.getUid))
+        .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(profile.providerType))
+    )
+    if (hasProvider) {
+      ctx
+        .update(AUTH_PROVIDER)
+        .set(AUTH_PROVIDER.PROVIDER_ID, profile.providerId)
+        .where(AUTH_PROVIDER.UID.eq(user.getUid))
+        .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(profile.providerType))
+        .execute()
+    } else {
+      authDao.insert(
+        new AuthProvider().tap { auth =>
+          auth.setUid(user.getUid)
+          auth.setProviderType(profile.providerType)
+          auth.setProviderId(profile.providerId)
+          auth.setCreatedAt(OffsetDateTime.now())
+        }
+      )
+    }
+  }
+}
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/auth/GoogleAuthResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/auth/GoogleAuthResource.scala
index aa0ca82a39c..6dd81c8cb0b 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/auth/GoogleAuthResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/auth/GoogleAuthResource.scala
@@ -19,101 +19,87 @@
 
 package org.apache.texera.web.resource.auth
 
-import com.google.api.client.googleapis.auth.oauth2.GoogleIdTokenVerifier
+import com.google.api.client.googleapis.auth.oauth2.{GoogleIdToken, GoogleIdTokenVerifier}
 import com.google.api.client.http.javanet.NetHttpTransport
 import com.google.api.client.json.gson.GsonFactory
-import org.apache.texera.auth.JwtAuth.{TOKEN_EXPIRE_TIME_IN_MINUTES, jwtClaims, jwtToken}
+import org.apache.texera.auth.JwtAuth.{jwtClaims, jwtToken}
 import org.apache.texera.common.config.UserSystemConfig
-import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
-import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
-import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.apache.texera.dao.jooq.generated.enums.ProviderTypeEnum
 import org.apache.texera.web.model.http.response.TokenIssueResponse
-import org.apache.texera.web.resource.auth.GoogleAuthResource.userDao
 
 import java.util.Collections
 import javax.ws.rs._
 import javax.ws.rs.core.MediaType
 
 object GoogleAuthResource {
-  private def userDao =
-    new UserDao(
-      SqlServer
-        .getInstance()
-        .createDSLContext()
-        .configuration
+
+  final private lazy val clientId = UserSystemConfig.googleClientId
+
+  /**
+    * Reduce a verified Google id-token payload to the fields we persist. Google omits `name`
+    * for accounts with no profile name, and the provisioner writes `name` straight to a NOT
+    * NULL column, so the address stands in for it. Only the last path segment of `picture` is
+    * kept — the frontend rebuilds the full `lh3.googleusercontent.com` URL around it.
+    *
+    * A payload with no address, or whose `email_verified` is not true, is refused rather than
+    * mapped — see [[ExternalProfile]] for why. Absent is not true: Google may omit the claim, and
+    * Workspace and custom-domain accounts can report false.
+    */
+  private[auth] def profileOf(payload: GoogleIdToken.Payload): ExternalProfile = {
+    val googleEmail = payload.getEmail
+    if (googleEmail == null || googleEmail.isBlank) {
+      throw new NotAuthorizedException("Login credentials are incorrect.")
+    }
+    if (!Option(payload.getEmailVerified).exists(_.booleanValue)) {
+      throw new NotAuthorizedException("Login credentials are incorrect.")
+    }
+    ExternalProfile(
+      ProviderTypeEnum.GOOGLE,
+      payload.getSubject,
+      Option(payload.get("name").asInstanceOf[String]).filter(_.nonEmpty).getOrElse(googleEmail),
+      googleEmail,
+      avatar = Option(payload.get("picture").asInstanceOf[String])
+        .flatMap(_.split("/").lastOption)
+        .getOrElse("")
     )
+  }
+
+  private lazy val verifier =
+    new GoogleIdTokenVerifier.Builder(new NetHttpTransport, GsonFactory.getDefaultInstance)
+      .setAudience(Collections.singletonList(clientId))
+      .build()
 }
 
 @Path("/auth/google")
 class GoogleAuthResource {
-  final private lazy val clientId = UserSystemConfig.googleClientId
 
   @GET
   @Path("/clientid")
-  def getClientId: String = clientId
+  def getClientId: String = GoogleAuthResource.clientId
+
+  /**
+    * Verify `credential` against Google, yielding its payload, or None if it is not a valid
+    * token for this client. This is the only seam that reaches the network, so tests override
+    * it instead of signing a token; it is kept as a method rather than a parameter because
+    * Jersey instantiates this resource from `classOf[GoogleAuthResource]`.
+    */
+  protected def verifiedPayload(credential: String): Option[GoogleIdToken.Payload] =
+    Option(GoogleAuthResource.verifier.verify(credential)).map(_.getPayload)
 
   @POST
   @Consumes(Array(MediaType.TEXT_PLAIN))
   @Produces(Array(MediaType.APPLICATION_JSON))
   @Path("/login")
-  def login(credential: String): TokenIssueResponse = {
-    val idToken =
-      new GoogleIdTokenVerifier.Builder(new NetHttpTransport, GsonFactory.getDefaultInstance)
-        .setAudience(
-          Collections.singletonList(clientId)
+  def login(credential: String): TokenIssueResponse =
+    verifiedPayload(credential) match {
+      case Some(payload) =>
+        val profile = GoogleAuthResource.profileOf(payload)
+        val user = ExternalAuthProvisioner.loginOrProvision(profile)
+        // The frontend reads `googleId` off the raw token; the provider id is already in hand
+        // here, so no lookup is needed.
+        TokenIssueResponse(
+          jwtToken(jwtClaims(user, Some(profile.providerId)))
         )
-        .build()
-        .verify(credential)
-    if (idToken != null) {
-      val payload = idToken.getPayload
-      val googleId = payload.getSubject
-      val googleName = payload.get("name").asInstanceOf[String]
-      val googleEmail = payload.getEmail
-      val googleAvatar = Option(payload.get("picture").asInstanceOf[String])
-        .flatMap(_.split("/").lastOption)
-        .getOrElse("")
-      val user = Option(userDao.fetchOneByGoogleId(googleId)) match {
-        case Some(user) =>
-          if (user.getName != googleName) {
-            user.setName(googleName)
-            userDao.update(user)
-          }
-          if (user.getEmail != googleEmail) {
-            user.setEmail(googleEmail)
-            userDao.update(user)
-          }
-          if (user.getGoogleAvatar != googleAvatar) {
-            user.setGoogleAvatar(googleAvatar)
-            userDao.update(user)
-          }
-          user
-        case None =>
-          Option(AuthResource.fetchUserByEmailIgnoreCase(googleEmail)) match {
-            case Some(user) =>
-              if (user.getName != googleName) {
-                user.setName(googleName)
-              }
-              user.setGoogleId(googleId)
-              user.setGoogleAvatar(googleAvatar)
-              if (user.getIsPlaceholder) {
-                AuthResource.claimPlaceholder(user)
-              }
-              userDao.update(user)
-              user
-            case None =>
-              // create a new user with googleId
-              val user = new User
-              user.setName(googleName)
-              user.setEmail(googleEmail)
-              user.setGoogleId(googleId)
-              user.setRole(UserRoleEnum.INACTIVE)
-              user.setGoogleAvatar(googleAvatar)
-              userDao.insert(user)
-              user
-          }
-      }
-      TokenIssueResponse(jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES)))
-    } else throw new NotAuthorizedException("Login credentials are incorrect.")
-  }
+      case None => throw new NotAuthorizedException("Login credentials are incorrect.")
+    }
 }
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/auth/LocalAuthProvisioner.scala b/amber/src/main/scala/org/apache/texera/web/resource/auth/LocalAuthProvisioner.scala
new file mode 100644
index 00000000000..c92f05658aa
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/resource/auth/LocalAuthProvisioner.scala
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.auth
+
+import org.apache.texera.dao.{SqlServer, SqlStates}
+import org.apache.texera.dao.jooq.generated.Tables.AUTH_PROVIDER
+import org.apache.texera.dao.jooq.generated.enums.ProviderTypeEnum
+import org.apache.texera.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
+import org.jasypt.util.password.StrongPasswordEncryptor
+import org.jooq.exception.DataAccessException
+
+import javax.ws.rs.WebApplicationException
+import javax.ws.rs.core.Response
+
+/**
+  * The LOCAL half of authentication: password hashing and the "insert a user together with the
+  * credential it logs in with" transaction. The counterpart to [[ExternalAuthProvisioner]].
+  *
+  * This exists because self-registration ([[AuthResource]]), admin-created accounts
+  * (`AdminUserResource`) and the admin bootstrap all need the same two-row insert, and each
+  * previously carried its own copy plus its own `StrongPasswordEncryptor`. A change to how a
+  * local credential is stored now lands in one place.
+  */
+object LocalAuthProvisioner {
+
+  private val passwordEncryptor = new StrongPasswordEncryptor
+
+  private def context = SqlServer.getInstance().context
+
+  def hashPassword(rawPassword: String): String =
+    passwordEncryptor.encryptPassword(rawPassword)
+
+  def checkPassword(rawPassword: String, hashedPassword: String): Boolean =
+    passwordEncryptor.checkPassword(rawPassword, hashedPassword)
+
+  /**
+    * Whether `handle` is already taken as a LOCAL login handle. Note this asks
+    * `auth_provider.provider_id`, not `"user".name` — the display name is mutable and is not
+    * identity, so it cannot answer this question.
+    */
+  def handleExists(handle: String): Boolean =
+    context.fetchExists(
+      context
+        .selectFrom(AUTH_PROVIDER)
+        .where(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+        .and(AUTH_PROVIDER.PROVIDER_ID.eq(handle))
+    )
+
+  /**
+    * Insert `user` and its LOCAL credential in one transaction, so a user row can never be left
+    * behind without the credential that makes it usable. `user` is mutated in place with the
+    * generated uid.
+    *
+    * The handle is passed explicitly rather than read off `user.getName`, so that identity is
+    * never re-derived from the mutable display name. Callers should pre-check with
+    * [[handleExists]] to report a friendly error; the unique-violation mapping here is the
+    * race fallback for two registrations of the same handle interleaving.
+    */
+  def createLocalAccount(user: User, handle: String, rawPassword: String): Unit = {
+    val hashedPassword = hashPassword(rawPassword)
+
+    try {
+      SqlServer.withTransaction(SqlServer.getInstance().createDSLContext()) { ctx =>
+        val txUserDao = new UserDao(ctx.configuration())
+        val txAuthDao = new AuthProviderDao(ctx.configuration())
+
+        txUserDao.insert(user)
+
+        val auth = new AuthProvider
+        auth.setUid(user.getUid)
+        auth.setProviderType(ProviderTypeEnum.LOCAL)
+        auth.setProviderId(handle)
+        auth.setPassword(hashedPassword)
+        txAuthDao.insert(auth)
+      }
+    } catch {
+      case e: DataAccessException if e.sqlState() == SqlStates.UNIQUE_VIOLATION =>
+        val message =
+          if (handleExists(handle)) s"Login handle $handle is already taken"
+          else s"Email ${user.getEmail} is already registered"
+        throw new WebApplicationException(message, e, Response.Status.CONFLICT)
+    }
+  }
+
+  /**
+    * Persist `user` and give it a LOCAL credential in one transaction, for an account row that
+    * already exists — claiming a dataset-contributor placeholder. The counterpart to
+    * [[createLocalAccount]], which inserts the user instead of updating it; both write the
+    * credential in the same transaction as the user row so an account can never be left in a
+    * state where it looks claimed but has nothing to log in with.
+    */
+  def claimWithLocalCredential(user: User, handle: String, rawPassword: String): Unit = {
+    val hashedPassword = hashPassword(rawPassword)
+
+    try {
+      SqlServer.withTransaction(SqlServer.getInstance().createDSLContext()) { ctx =>
+        new UserDao(ctx.configuration()).update(user)
+
+        val auth = new AuthProvider
+        auth.setUid(user.getUid)
+        auth.setProviderType(ProviderTypeEnum.LOCAL)
+        auth.setProviderId(handle)
+        auth.setPassword(hashedPassword)
+        new AuthProviderDao(ctx.configuration()).insert(auth)
+      }
+    } catch {
+      case e: DataAccessException if e.sqlState() == SqlStates.UNIQUE_VIOLATION =>
+        val message =
+          if (handleExists(handle)) s"Login handle $handle is already taken"
+          else s"Account for ${user.getEmail} has already been claimed"
+        throw new WebApplicationException(message, e, Response.Status.CONFLICT)
+    }
+  }
+}
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/DashboardResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/DashboardResource.scala
index 3d78eef3035..795a19ff109 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/DashboardResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/DashboardResource.scala
@@ -221,7 +221,7 @@ class DashboardResource {
     val scalaUserIds: Set[Integer] = userIds.asScala.toSet
 
     val records = context
-      .select(USER.UID, USER.NAME, USER.GOOGLE_AVATAR)
+      .select(USER.UID, USER.NAME, USER.AVATAR)
       .from(USER)
       .where(USER.UID.in(scalaUserIds.asJava))
       .fetch()
@@ -230,7 +230,7 @@ class DashboardResource {
       .map { record =>
         val userId = record.get(USER.UID)
         val userName = record.get(USER.NAME)
-        val googleAvatar = Option(record.get(USER.GOOGLE_AVATAR))
+        val googleAvatar = Option(record.get(USER.AVATAR))
         userId -> UserInfo(userId, userName, googleAvatar)
       }
       .toMap
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResource.scala
index cb426f787b2..b550ac46e00 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResource.scala
@@ -20,19 +20,21 @@
 package org.apache.texera.web.resource.dashboard.admin.user
 
 import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
 import org.apache.texera.dao.jooq.generated.tables.User.USER
+import org.apache.texera.dao.jooq.generated.tables.AuthProvider.AUTH_PROVIDER
 import org.apache.texera.dao.jooq.generated.tables.UserLastActiveTime.USER_LAST_ACTIVE_TIME
 import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
 import org.apache.texera.dao.jooq.generated.tables.pojos.User
 import org.apache.texera.web.resource.EmailTemplate.createRoleChangeTemplate
 import org.apache.texera.web.resource.GmailResource.sendEmail
+import org.apache.texera.web.resource.auth.LocalAuthProvisioner
 import org.apache.texera.web.resource.dashboard.admin.user.AdminUserResource.userDao
 import org.apache.texera.web.resource.dashboard.user.dataset.utils.DatasetStatisticsUtils.getUserCreatedDatasets
 import org.apache.texera.web.resource.dashboard.user.quota.UserQuotaResource._
-import org.jasypt.util.password.StrongPasswordEncryptor
 
 import java.util
+import java.util.UUID
 import javax.annotation.security.RolesAllowed
 import javax.ws.rs._
 import javax.ws.rs.core.{MediaType, Response}
@@ -43,6 +45,8 @@ case class UserInfo(
     email: String,
     googleId: String,
     role: UserRoleEnum,
+    // `"user".avatar` is no longer Google-specific, but this is the JSON key
+    // `admin-user.component.html` binds, so the wire name stays until the frontend migrates.
     googleAvatar: String,
     comment: String,
     lastLogin: java.time.OffsetDateTime, // will be null if never logged in
@@ -73,14 +77,23 @@ class AdminUserResource {
   @Path("/list")
   @Produces(Array(MediaType.APPLICATION_JSON))
   def list(): util.List[UserInfo] = {
+
+    val googleProvider = AUTH_PROVIDER.as("google_provider")
+
     AdminUserResource.context
       .select(
         USER.UID,
         USER.NAME,
         USER.EMAIL,
-        USER.GOOGLE_ID,
+        // fetchInto maps onto a Scala case class POSITIONALLY, not by name: a case class has no
+        // no-arg constructor, so jOOQ falls through to ImmutablePOJOMapper. So the column order
+        // below must track the UserInfo field order — adding, removing or reordering a projected
+        // column here without doing the same to UserInfo silently shifts every later field.
+        // `last_active_time` landing on `lastLogin` only works because of that. The aliases are
+        // documentation; they do not drive the mapping.
+        googleProvider.PROVIDER_ID.as("googleId"),
         USER.ROLE,
-        USER.GOOGLE_AVATAR,
+        USER.AVATAR.as("googleAvatar"),
         USER.COMMENT,
         USER_LAST_ACTIVE_TIME.LAST_ACTIVE_TIME,
         USER.ACCOUNT_CREATION_TIME,
@@ -91,6 +104,9 @@ class AdminUserResource {
       .from(USER)
       .leftJoin(USER_LAST_ACTIVE_TIME)
       .on(USER.UID.eq(USER_LAST_ACTIVE_TIME.UID))
+      .leftJoin(googleProvider)
+      .on(googleProvider.PROVIDER_TYPE.eq(ProviderTypeEnum.GOOGLE))
+      .and(googleProvider.UID.eq(USER.UID))
       .fetchInto(classOf[UserInfo])
   }
 
@@ -119,12 +135,13 @@ class AdminUserResource {
   @POST
   @Path("/add")
   def addUser(): Unit = {
-    val random = System.currentTimeMillis().toString
-    val newUser = new User
-    newUser.setName("User" + random)
-    newUser.setPassword(new StrongPasswordEncryptor().encryptPassword(random))
-    newUser.setRole(UserRoleEnum.INACTIVE)
-    userDao.insert(newUser)
+    // Two independent UUIDs: the handle is visible to anyone who can read /list, so deriving the
+    // password from it would let any such caller log in as the new account.
+    val handle = "User" + UUID.randomUUID().toString
+    val user = new User
+    user.setName(handle)
+    user.setRole(UserRoleEnum.INACTIVE)
+    LocalAuthProvisioner.createLocalAccount(user, handle, UUID.randomUUID().toString)
   }
 
   @GET
diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
index d1946f8d727..cca18443b7d 100644
--- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
+++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala
@@ -321,7 +321,7 @@ object WorkflowExecutionsResource {
         WORKFLOW_EXECUTIONS.VID,
         WORKFLOW_EXECUTIONS.CUID,
         USER.NAME,
-        USER.GOOGLE_AVATAR,
+        USER.AVATAR,
         WORKFLOW_EXECUTIONS.STATUS,
         WORKFLOW_EXECUTIONS.RESULT,
         WORKFLOW_EXECUTIONS.STARTING_TIME,
@@ -582,7 +582,7 @@ class WorkflowExecutionsResource {
             WORKFLOW_EXECUTIONS.VID,
             WORKFLOW_EXECUTIONS.CUID,
             USER.NAME,
-            USER.GOOGLE_AVATAR,
+            USER.AVATAR,
             WORKFLOW_EXECUTIONS.STATUS,
             WORKFLOW_EXECUTIONS.RESULT,
             WORKFLOW_EXECUTIONS.STARTING_TIME,
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala b/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
index 605e6a9e582..4e73a0e655f 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ResultExportService.scala
@@ -36,7 +36,7 @@ import org.apache.arrow.vector.ipc.ArrowFileWriter
 import org.apache.commons.io.IOUtils
 import org.apache.commons.lang3.StringUtils
 import org.apache.texera.auth.JwtAuth
-import org.apache.texera.auth.JwtAuth.{TOKEN_EXPIRE_TIME_IN_MINUTES, jwtClaims}
+import org.apache.texera.auth.JwtAuth.jwtClaims
 import org.apache.texera.dao.jooq.generated.tables.pojos.User
 import org.apache.texera.web.model.http.request.result.{OperatorExportInfo, ResultExportRequest}
 import org.apache.texera.web.model.http.response.result.ResultExportResponse
@@ -550,7 +550,7 @@ class ResultExportService(workflowIdentity: WorkflowIdentity, computingUnitId: I
         connection.setRequestProperty("Content-Type", "application/octet-stream")
         connection.setRequestProperty(
           "Authorization",
-          s"Bearer ${JwtAuth.jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES))}"
+          s"Bearer ${JwtAuth.jwtToken(jwtClaims(user))}"
         )
         connection.setChunkedStreamingMode(0)
 
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala
index 595a9b07c91..607330fd155 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/DefaultCostEstimatorSpec.scala
@@ -65,7 +65,6 @@ class DefaultCostEstimatorSpec
     user.setUid(Integer.valueOf(1))
     user.setName("test_user")
     user.setRole(UserRoleEnum.ADMIN)
-    user.setPassword("123")
     user.setEmail("test_user@test.com")
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
index ac3bf167d38..f5c4657b8e6 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/TestUtils.scala
@@ -260,7 +260,6 @@ object TestUtils {
     user.setUid(Integer.valueOf(id))
     user.setName(s"test_user_$id")
     user.setRole(UserRoleEnum.ADMIN)
-    user.setPassword("123")
     user.setEmail(s"test_user_$id@test.com")
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/auth/UserAuthenticatorSpec.scala b/amber/src/test/scala/org/apache/texera/web/auth/UserAuthenticatorSpec.scala
index d185669caaf..6c62f9fa2df 100644
--- a/amber/src/test/scala/org/apache/texera/web/auth/UserAuthenticatorSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/auth/UserAuthenticatorSpec.scala
@@ -35,7 +35,6 @@ class UserAuthenticatorSpec extends AnyFlatSpec with Matchers {
     val claims = new JwtClaims
     claims.setSubject("alice")
     claims.setClaim("userId", 42)
-    claims.setClaim("googleId", "g-123")
     claims.setClaim("email", "alice@example.com")
     claims.setClaim("role", UserRoleEnum.ADMIN.name)
     claims.setClaim("googleAvatar", "avatar-blob")
@@ -55,8 +54,7 @@ class UserAuthenticatorSpec extends AnyFlatSpec with Matchers {
     u.getUid shouldBe 42
     u.getName shouldBe "alice"
     u.getEmail shouldBe "alice@example.com"
-    u.getGoogleId shouldBe "g-123"
-    u.getGoogleAvatar shouldBe "avatar-blob"
+    u.getAvatar shouldBe "avatar-blob"
     u.getRole shouldBe UserRoleEnum.ADMIN
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala
index 96f3aae8483..a313a667a27 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/CollaborationResourceSpec.scala
@@ -96,7 +96,6 @@ class CollaborationResourceSpec
     user.setUid(Integer.valueOf(accessUid))
     user.setName("collab_lock_user")
     user.setRole(UserRoleEnum.REGULAR)
-    user.setPassword("pw")
     new UserDao(getDSLContext.configuration()).insert(user)
 
     val workflow = new Workflow
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala
index 10e1e5b357d..d18387ee8ce 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/FeedbackResourceSpec.scala
@@ -50,7 +50,6 @@ class FeedbackResourceSpec
     user.setUid(uid)
     user.setName(name)
     user.setEmail(s"user_${UUID.randomUUID()}@example.com")
-    user.setPassword("password")
     user.setRole(UserRoleEnum.REGULAR)
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/UserConfigResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/UserConfigResourceSpec.scala
index 0ee7158077c..1fd49ab3992 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/UserConfigResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/UserConfigResourceSpec.scala
@@ -67,7 +67,6 @@ class UserConfigResourceSpec
     user.setUid(uid)
     user.setName(name)
     user.setEmail(email)
-    user.setPassword("password")
     userDao.insert(user)
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala
index d91a1dc1afe..d323d4c9757 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala
@@ -22,10 +22,10 @@ package org.apache.texera.web.resource.auth
 import org.apache.texera.auth.JwtAuth
 import org.apache.texera.common.config.UserSystemConfig
 import org.apache.texera.dao.MockTexeraDB
-import org.apache.texera.dao.jooq.generated.Tables.USER
-import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
-import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
-import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
 import org.apache.texera.web.model.http.request.auth.{UserLoginRequest, UserRegistrationRequest}
 import org.jasypt.util.password.StrongPasswordEncryptor
 import org.scalatest.flatspec.AnyFlatSpec
@@ -49,6 +49,7 @@ class AuthResourceSpec
   private val encryptor = new StrongPasswordEncryptor()
 
   private var userDao: UserDao = _
+  private var authDao: AuthProviderDao = _
   private var resource: AuthResource = _
 
   private def uname(tag: String): String = s"authspec_${tag}_$runId"
@@ -61,12 +62,14 @@ class AuthResourceSpec
 
   override protected def beforeEach(): Unit = {
     userDao = new UserDao(getDSLContext.configuration())
+    authDao = new AuthProviderDao(getDSLContext.configuration())
     resource = new AuthResource()
     cleanup()
   }
 
   override protected def afterEach(): Unit = cleanup()
 
+  // The auth_provider FK is ON DELETE CASCADE, so deleting the user clears its credential rows.
   private def cleanup(): Unit = {
     // startsWith escapes SQL LIKE wildcards, so the literal "authspec_" prefix is matched exactly.
     getDSLContext.deleteFrom(USER).where(USER.NAME.startsWith("authspec_")).execute()
@@ -74,6 +77,10 @@ class AuthResourceSpec
     getDSLContext.deleteFrom(USER).where(USER.NAME.eq(UserSystemConfig.adminUsername)).execute()
   }
 
+  /**
+    * Seed a user plus the LOCAL auth_provider row it logs in with, mirroring
+    * `LocalAuthProvisioner.createLocalAccount`.
+    */
   private def seedUser(
       name: String,
       password: String,
@@ -82,12 +89,37 @@ class AuthResourceSpec
     val user = new User
     user.setName(name)
     user.setEmail(s"$name@example.com")
-    user.setPassword(encryptor.encryptPassword(password))
     user.setRole(role)
     userDao.insert(user)
+
+    val auth = new AuthProvider
+    auth.setUid(user.getUid)
+    auth.setProviderType(ProviderTypeEnum.LOCAL)
+    // The login handle is the provider id, not the (mutable) display name.
+    auth.setProviderId(name)
+    auth.setPassword(encryptor.encryptPassword(password))
+    authDao.insert(auth)
     user
   }
 
+  /** The external id `uid` authenticates with at `providerType`, if it has one. */
+  private def providerIdOf(uid: Integer, providerType: ProviderTypeEnum): String =
+    getDSLContext
+      .select(AUTH_PROVIDER.PROVIDER_ID)
+      .from(AUTH_PROVIDER)
+      .where(AUTH_PROVIDER.UID.eq(uid))
+      .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(providerType))
+      .fetchOne(AUTH_PROVIDER.PROVIDER_ID)
+
+  /** The stored LOCAL password hash for a login handle. */
+  private def storedPasswordOf(handle: String): String =
+    getDSLContext
+      .select(AUTH_PROVIDER.PASSWORD)
+      .from(AUTH_PROVIDER)
+      .where(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+      .and(AUTH_PROVIDER.PROVIDER_ID.eq(handle))
+      .fetchOne(AUTH_PROVIDER.PASSWORD)
+
   private def subjectOf(token: String): String =
     JwtAuth.jwtConsumer.processToClaims(token).getSubject
 
@@ -142,8 +174,9 @@ class AuthResourceSpec
     stored.getEmail shouldBe uemail("reg")
     stored.getIsPlaceholder shouldBe false
     // stored hashed, not in plain text, but verifies against the plain password
-    stored.getPassword should not be "pw"
-    encryptor.checkPassword("pw", stored.getPassword) shouldBe true
+    val storedPassword = storedPasswordOf(uname("reg"))
+    storedPassword should not be "pw"
+    encryptor.checkPassword("pw", storedPassword) shouldBe true
   }
 
   it should "reject an empty username" in {
@@ -204,6 +237,31 @@ class AuthResourceSpec
     user
   }
 
+  /**
+    * Seed an external (non-LOCAL) credential for an existing user. ck_provider_credential
+    * requires a password for LOCAL and only for LOCAL, so this leaves it null.
+    */
+  private def seedExternalProvider(
+      uid: Integer,
+      providerType: ProviderTypeEnum,
+      providerId: String
+  ): Unit = {
+    val auth = new AuthProvider
+    auth.setUid(uid)
+    auth.setProviderType(providerType)
+    auth.setProviderId(providerId)
+    authDao.insert(auth)
+  }
+
+  /** Whether `uid` holds a credential of the given kind. */
+  private def hasProvider(uid: Integer, providerType: ProviderTypeEnum): Boolean =
+    getDSLContext.fetchExists(
+      getDSLContext
+        .selectFrom(AUTH_PROVIDER)
+        .where(AUTH_PROVIDER.UID.eq(uid))
+        .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(providerType))
+    )
+
   it should "claim a placeholder account with the matching email" in {
     val placeholder = seedPlaceholder(uname("ghost"), uemail("claim"))
 
@@ -214,7 +272,8 @@ class AuthResourceSpec
     val claimed = userDao.fetchOneByEmail(uemail("claim"))
     claimed.getUid shouldEqual placeholder.getUid
     claimed.getIsPlaceholder shouldBe false
-    encryptor.checkPassword("secret-pw", claimed.getPassword) shouldBe true
+    // the credential lives in auth_provider, keyed by the login handle just registered
+    encryptor.checkPassword("secret-pw", storedPasswordOf(uname("claimer"))) shouldBe true
     claimed.getRole shouldEqual UserRoleEnum.INACTIVE
     claimed.getComment should include("Claimed contributor placeholder at ")
   }
@@ -233,9 +292,9 @@ class AuthResourceSpec
     val real = new User
     real.setName(uname("real"))
     real.setEmail(uemail("real"))
-    real.setGoogleId(s"google-$runId")
     real.setRole(UserRoleEnum.INACTIVE)
     userDao.insert(real)
+    seedExternalProvider(real.getUid, ProviderTypeEnum.GOOGLE, s"google-$runId")
 
     val ex = intercept[NotAcceptableException](
       resource.register(UserRegistrationRequest(uname("attacker"), uemail("real"), "attacker-pw"))
@@ -243,9 +302,10 @@ class AuthResourceSpec
     ex.getMessage should include("Email exists")
 
     val untouched = userDao.fetchOneByEmail(uemail("real"))
-    untouched.getPassword shouldBe null
-    untouched.getGoogleId shouldEqual s"google-$runId"
     untouched.getIsPlaceholder shouldBe false
+    // no LOCAL credential was grafted on, and the Google identity is intact
+    hasProvider(untouched.getUid, ProviderTypeEnum.LOCAL) shouldBe false
+    providerIdOf(untouched.getUid, ProviderTypeEnum.GOOGLE) shouldEqual s"google-$runId"
   }
 
   it should "reject claiming with an already-taken username" in {
@@ -273,6 +333,32 @@ class AuthResourceSpec
     ex.getMessage should include("Email exists")
   }
 
+  // The taken-handle check has to ask auth_provider, not "user".name: an external login rewrites
+  // the display name but never the handle, so a name-based guard let the handle through and the
+  // insert then died on uq_provider_identity as a 500 instead of this 406.
+  it should "reject a taken handle even when the owner's display name has since drifted" in {
+    val owner = seedUser(uname("drift"), "pw")
+    // keep the cleanup prefix so the renamed row is still collected
+    owner.setName(uname("drift_renamed_by_google"))
+    userDao.update(owner)
+
+    val ex = intercept[NotAcceptableException](
+      resource.register(UserRegistrationRequest(uname("drift"), uemail("drift_other"), "pw2"))
+    )
+    ex.getMessage should include("Username exists")
+  }
+
+  // The mirror case: a display name that is not a handle must not block registration.
+  it should "allow a handle that only collides with some other account's display name" in {
+    val squatter = seedUser(uname("squatter"), "pw")
+    squatter.setName(uname("wanted"))
+    userDao.update(squatter)
+
+    val response =
+      resource.register(UserRegistrationRequest(uname("wanted"), uemail("wanted"), "pw2"))
+    subjectOf(response.accessToken) shouldBe uname("wanted")
+  }
+
   // ─── createAdminUser ────────────────────────────────────────────────────────
 
   "createAdminUser" should "insert the configured admin with the ADMIN role and a hashed password" in {
@@ -281,7 +367,10 @@ class AuthResourceSpec
     val admins = userDao.fetchByName(UserSystemConfig.adminUsername)
     admins.size() shouldBe 1
     admins.get(0).getRole shouldBe UserRoleEnum.ADMIN
-    encryptor.checkPassword(UserSystemConfig.adminPassword, admins.get(0).getPassword) shouldBe true
+    encryptor.checkPassword(
+      UserSystemConfig.adminPassword,
+      storedPasswordOf(UserSystemConfig.adminUsername)
+    ) shouldBe true
   }
 
   it should "not create a second admin when one already exists" in {
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisionerSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisionerSpec.scala
new file mode 100644
index 00000000000..44b2efe683b
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisionerSpec.scala
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.auth
+
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
+import org.jooq.impl.DSL
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+/**
+  * Integration spec for [[ExternalAuthProvisioner]] against embedded Postgres
+  * ([[MockTexeraDB]] loads the real `texera_ddl.sql`, so the `auth_provider` table and
+  * its `ck_provider_credential` / `uq_provider_identity` constraints are exercised).
+  * `loginOrProvision` runs the same transaction the Google resources call.
+  */
+class ExternalAuthProvisionerSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with MockTexeraDB {
+
+  // All test users share this email suffix so cleanup can target them precisely;
+  // the auth_provider FK is ON DELETE CASCADE, so deleting the user clears its rows.
+  private val emailDomain = "@provisioner-test.com"
+
+  private var userDao: UserDao = _
+  private var authDao: AuthProviderDao = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+    authDao = new AuthProviderDao(getDSLContext.configuration())
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  override protected def beforeEach(): Unit = cleanup()
+  override protected def afterEach(): Unit = cleanup()
+
+  // Case-insensitive so it also collects rows seeded with a differing casing.
+  private def cleanup(): Unit =
+    getDSLContext.deleteFrom(USER).where(DSL.lower(USER.EMAIL).like("%" + emailDomain)).execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  private def profile(
+      providerId: String,
+      name: String,
+      email: String,
+      avatar: String = "pic"
+  ): ExternalProfile =
+    ExternalProfile(ProviderTypeEnum.GOOGLE, providerId, name, email, avatar)
+
+  /** Seed a user row directly; uid is DB-assigned and read back into the pojo. */
+  private def seedUser(name: String, localPart: String, avatar: String = null): User =
+    seedUserWithEmail(name, localPart + emailDomain, avatar)
+
+  /** Seed a user row at a verbatim address, for the casing tests. */
+  private def seedUserWithEmail(name: String, email: String, avatar: String = null): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(email)
+    user.setRole(UserRoleEnum.REGULAR)
+    if (avatar != null) user.setAvatar(avatar)
+    userDao.insert(user)
+    user
+  }
+
+  /** Seed an external (non-LOCAL) provider row for an existing user. */
+  private def seedExternalProvider(uid: Integer, pt: ProviderTypeEnum, providerId: String): Unit = {
+    val auth = new AuthProvider
+    auth.setUid(uid)
+    auth.setProviderType(pt)
+    auth.setProviderId(providerId)
+    authDao.insert(auth)
+  }
+
+  private def providerRowCount(uid: Integer): Int =
+    getDSLContext.fetchCount(AUTH_PROVIDER, AUTH_PROVIDER.UID.eq(uid))
+
+  private def providerIdOf(uid: Integer, pt: ProviderTypeEnum): String =
+    getDSLContext
+      .select(AUTH_PROVIDER.PROVIDER_ID)
+      .from(AUTH_PROVIDER)
+      .where(AUTH_PROVIDER.UID.eq(uid))
+      .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(pt))
+      .fetchOne(AUTH_PROVIDER.PROVIDER_ID)
+
+  /** Counts case-insensitively, so a duplicate differing only in case is still counted. */
+  private def userCountByEmail(localPart: String): Int =
+    getDSLContext.fetchCount(USER, DSL.lower(USER.EMAIL).eq(localPart + emailDomain))
+
+  // ---- new-identity provisioning -------------------------------------------
+
+  "ExternalAuthProvisioner.loginOrProvision" should "create an INACTIVE user and provider row for a brand-new Google identity" in {
+    val user = ExternalAuthProvisioner.loginOrProvision(
+      profile("google-sub-1", "New User", "new" + emailDomain, avatar = "avatar1")
+    )
+
+    user.getUid should not be null
+    user.getName shouldBe "New User"
+    user.getEmail shouldBe "new" + emailDomain
+    user.getAvatar shouldBe "avatar1"
+    user.getRole shouldBe UserRoleEnum.INACTIVE
+
+    providerRowCount(user.getUid) shouldBe 1
+    providerIdOf(user.getUid, ProviderTypeEnum.GOOGLE) shouldBe "google-sub-1"
+  }
+
+  // ---- returning known identity --------------------------------------------
+
+  it should "be idempotent for a returning identity (same uid, no duplicate provider row or user)" in {
+    val p = profile("google-sub-return", "Ret", "ret" + emailDomain, avatar = "a")
+
+    val first = ExternalAuthProvisioner.loginOrProvision(p)
+    val second = ExternalAuthProvisioner.loginOrProvision(p)
+
+    second.getUid shouldBe first.getUid
+    providerRowCount(first.getUid) shouldBe 1
+    userCountByEmail("ret") shouldBe 1
+  }
+
+  it should "refresh drifted profile fields for a known identity" in {
+    ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-drift", "Old Name", "drift" + emailDomain, avatar = "oldpic")
+    )
+    val updated = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-drift", "New Name", "drift" + emailDomain, avatar = "newpic")
+    )
+
+    updated.getName shouldBe "New Name"
+    updated.getAvatar shouldBe "newpic"
+    // confirm it persisted, not just mutated in memory
+    userDao.fetchOneByUid(updated.getUid).getName shouldBe "New Name"
+    userDao.fetchOneByUid(updated.getUid).getAvatar shouldBe "newpic"
+  }
+
+  it should "adopt the provider's new email address for a known identity" in {
+    val created = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-rename", "Renamer", "before" + emailDomain)
+    )
+
+    val updated = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-rename", "Renamer", "after" + emailDomain)
+    )
+
+    updated.getUid shouldBe created.getUid
+    userDao.fetchOneByUid(created.getUid).getEmail shouldBe "after" + emailDomain
+    userCountByEmail("before") shouldBe 0
+  }
+
+  // ---- email match, no provider yet ----------------------------------------
+
+  it should "link a new provider to an existing email-matched user instead of creating a duplicate" in {
+    val existing = seedUser("Local User", "linkme")
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-link", "Local User", "linkme" + emailDomain)
+    )
+
+    result.getUid shouldBe existing.getUid
+    userCountByEmail("linkme") shouldBe 1
+    providerIdOf(existing.getUid, ProviderTypeEnum.GOOGLE) shouldBe "sub-link"
+  }
+
+  // `"user".email` is a plain case-sensitive UNIQUE and `idx_user_email_lower` is not unique, so
+  // an exact-match lookup would not merely miss — the follow-up insert would succeed and fork the
+  // account silently. Registration stores the address as typed while contributor placeholders are
+  // stored lower-cased, so the casings really do differ in practice. Mirrors the register-path
+  // guard asserted in AuthResourceSpec.
+  it should "link to an existing account whose stored email differs only in case" in {
+    val existing = seedUserWithEmail("Mixed Case", "MixedCase" + emailDomain)
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-casing", "Mixed Case", "mixedcase" + emailDomain)
+    )
+
+    result.getUid shouldBe existing.getUid
+    userCountByEmail("mixedcase") shouldBe 1
+    providerIdOf(existing.getUid, ProviderTypeEnum.GOOGLE) shouldBe "sub-casing"
+  }
+
+  // A contributor placeholder is stored lower-cased by DatasetResource, so a provider reporting
+  // the address with different casing must still claim it rather than orphan the contributor link.
+  it should "claim a placeholder account whose stored email differs only in case" in {
+    val placeholder = seedUserWithEmail("Ghost", "ghost" + emailDomain)
+    placeholder.setIsPlaceholder(true)
+    userDao.update(placeholder)
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-ghost", "Ghost", "GHOST" + emailDomain)
+    )
+
+    result.getUid shouldBe placeholder.getUid
+    userDao.fetchOneByUid(placeholder.getUid).getIsPlaceholder shouldBe false
+    userCountByEmail("ghost") shouldBe 1
+  }
+
+  it should "claim a placeholder account when an external identity presents its email" in {
+    val placeholder = seedUser("Placeholder", "claimme")
+    placeholder.setIsPlaceholder(true)
+    userDao.update(placeholder)
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      profile("sub-claim", "Claimer", "claimme" + emailDomain)
+    )
+
+    result.getUid shouldBe placeholder.getUid
+    val claimed = userDao.fetchOneByUid(placeholder.getUid)
+    claimed.getIsPlaceholder shouldBe false
+    claimed.getComment should include("Claimed contributor placeholder at ")
+  }
+
+  // ---- provider id rotation -------------------------------------------------
+
+  it should "update the stored provider id when the same user returns with a new one" in {
+    val existing = seedUser("Rotating", "rotate")
+    seedExternalProvider(existing.getUid, ProviderTypeEnum.GOOGLE, "old-sub")
+
+    ExternalAuthProvisioner.loginOrProvision(
+      profile("new-sub", "Rotating", "rotate" + emailDomain)
+    )
+
+    providerRowCount(existing.getUid) shouldBe 1
+    providerIdOf(existing.getUid, ProviderTypeEnum.GOOGLE) shouldBe "new-sub"
+  }
+}
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/auth/GoogleAuthResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/auth/GoogleAuthResourceSpec.scala
new file mode 100644
index 00000000000..02aed679514
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/GoogleAuthResourceSpec.scala
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.auth
+
+import com.google.api.client.googleapis.auth.oauth2.GoogleIdToken
+import org.apache.texera.common.config.UserSystemConfig
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.UserDao
+import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import javax.ws.rs.NotAuthorizedException
+
+/**
+  * Integration spec for [[GoogleAuthResource]] against embedded Postgres.
+  *
+  * Token verification is the one part that cannot run here — it needs a Google-signed JWT and a
+  * network round trip — so the suite overrides `verifiedPayload` and drives the resource with
+  * payloads built by hand. What it pins down is everything downstream of verification: how a
+  * Google payload becomes an [[ExternalProfile]] (the name fallback and the avatar reduced to
+  * its last path segment) and that a credential Google does not verify is a 401.
+  */
+class GoogleAuthResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with MockTexeraDB {
+
+  private val emailDomain = "@google-auth-test.com"
+
+  private var userDao: UserDao = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  override protected def beforeEach(): Unit = cleanup()
+  override protected def afterEach(): Unit = cleanup()
+
+  private def cleanup(): Unit =
+    getDSLContext.deleteFrom(USER).where(USER.EMAIL.like("%" + emailDomain)).execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  /** A resource whose verification step always yields `payload`, standing in for Google. */
+  private class StubbedGoogleAuthResource(payload: Option[GoogleIdToken.Payload])
+      extends GoogleAuthResource {
+    override protected def verifiedPayload(credential: String): Option[GoogleIdToken.Payload] =
+      payload
+  }
+
+  /**
+    * A payload shaped like Google's. `name` and `picture` are ordinary JSON members rather than
+    * typed fields, and passing null omits them — which is exactly the case the name fallback
+    * exists for. `emailVerified` defaults to true, the only kind of payload the resource accepts;
+    * null omits the claim.
+    */
+  private def payload(
+      subject: String,
+      email: String,
+      name: String = "Given Name",
+      picture: String = "https://lh3.googleusercontent.com/a/AVATAR-ID",
+      emailVerified: java.lang.Boolean = true
+  ): GoogleIdToken.Payload = {
+    val p = new GoogleIdToken.Payload()
+    p.setSubject(subject)
+    p.setEmail(email)
+    if (name != null) p.set("name", name)
+    if (picture != null) p.set("picture", picture)
+    if (emailVerified != null) p.setEmailVerified(emailVerified)
+    p
+  }
+
+  private def loginWith(p: GoogleIdToken.Payload): Unit =
+    new StubbedGoogleAuthResource(Some(p))
+      .login("stubbed-credential")
+      .accessToken should not be empty
+
+  private def userByEmail(localPart: String): User =
+    userDao.fetchOneByEmail(localPart + emailDomain)
+
+  private def googleIdOf(uid: Integer): String =
+    getDSLContext
+      .select(AUTH_PROVIDER.PROVIDER_ID)
+      .from(AUTH_PROVIDER)
+      .where(AUTH_PROVIDER.UID.eq(uid))
+      .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.GOOGLE))
+      .fetchOne(AUTH_PROVIDER.PROVIDER_ID)
+
+  // ---- first login ---------------------------------------------------------
+
+  behavior of "login"
+
+  it should "provision an INACTIVE user and a GOOGLE provider row on a first login" in {
+    loginWith(payload("google-sub-new", "newcomer" + emailDomain, name = "New Comer"))
+
+    val user = userByEmail("newcomer")
+    user should not be null
+    user.getName shouldBe "New Comer"
+    user.getRole shouldBe UserRoleEnum.INACTIVE
+    googleIdOf(user.getUid) shouldBe "google-sub-new"
+  }
+
+  it should "return the same account on a second login rather than provisioning again" in {
+    loginWith(payload("google-sub-repeat", "repeat" + emailDomain))
+    val first = userByEmail("repeat").getUid
+
+    loginWith(payload("google-sub-repeat", "repeat" + emailDomain))
+
+    getDSLContext.fetchCount(USER, USER.EMAIL.eq("repeat" + emailDomain)) shouldBe 1
+    userByEmail("repeat").getUid shouldBe first
+  }
+
+  // ---- payload mapping -----------------------------------------------------
+
+  it should "fall back to the email address when the payload carries no name" in {
+    loginWith(payload("google-sub-nameless", "nameless" + emailDomain, name = null))
+
+    userByEmail("nameless").getName shouldBe "nameless" + emailDomain
+  }
+
+  it should "fall back to the email address when the name is present but blank" in {
+    loginWith(payload("google-sub-blank", "blank" + emailDomain, name = ""))
+
+    userByEmail("blank").getName shouldBe "blank" + emailDomain
+  }
+
+  it should "store only the last path segment of the picture URL" in {
+    loginWith(payload("google-sub-avatar", "avatar" + emailDomain))
+
+    userByEmail("avatar").getAvatar shouldBe "AVATAR-ID"
+  }
+
+  it should "store an empty avatar when the payload carries no picture" in {
+    loginWith(payload("google-sub-nopic", "nopic" + emailDomain, picture = null))
+
+    userByEmail("nopic").getAvatar shouldBe ""
+  }
+
+  // ---- verification failure ------------------------------------------------
+
+  it should "reject a credential Google does not verify with a 401" in {
+    val resource = new StubbedGoogleAuthResource(None)
+
+    a[NotAuthorizedException] should be thrownBy resource.login("not-a-real-credential")
+  }
+
+  // Not merely untidy input: matching on an unverified address is an account takeover.
+  it should "reject a token whose email_verified is false, leaving the matching account alone" in {
+    loginWith(payload("google-sub-owner", "victim" + emailDomain, name = "Real Owner"))
+    val owner = userByEmail("victim").getUid
+
+    val resource = new StubbedGoogleAuthResource(
+      Some(payload("google-sub-attacker", "victim" + emailDomain, emailVerified = false))
+    )
+    a[NotAuthorizedException] should be thrownBy resource.login("stubbed-credential")
+
+    // the victim's account still points at the original identity, and no second one was added
+    googleIdOf(owner) shouldBe "google-sub-owner"
+    getDSLContext.fetchCount(USER, USER.EMAIL.eq("victim" + emailDomain)) shouldBe 1
+  }
+
+  it should "reject a token that omits email_verified rather than assuming it" in {
+    val resource = new StubbedGoogleAuthResource(
+      Some(payload("google-sub-noflag", "noflag" + emailDomain, emailVerified = null))
+    )
+
+    a[NotAuthorizedException] should be thrownBy resource.login("stubbed-credential")
+    userByEmail("noflag") shouldBe null
+  }
+
+  // A null address NPEs in `EmailUtil.normalize`, and as the name fallback violates NOT NULL.
+  it should "reject a token with no email address" in {
+    val resource = new StubbedGoogleAuthResource(Some(payload("google-sub-noemail", null)))
+
+    a[NotAuthorizedException] should be thrownBy resource.login("stubbed-credential")
+  }
+
+  it should "reject a token whose email is blank" in {
+    val resource = new StubbedGoogleAuthResource(Some(payload("google-sub-blankemail", "   ")))
+
+    a[NotAuthorizedException] should be thrownBy resource.login("stubbed-credential")
+  }
+
+  // ---- client id -----------------------------------------------------------
+
+  behavior of "getClientId"
+
+  it should "expose the configured Google client id" in {
+    new GoogleAuthResource().getClientId shouldBe UserSystemConfig.googleClientId
+  }
+}
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/auth/LocalAuthProvisionerSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/auth/LocalAuthProvisionerSpec.scala
new file mode 100644
index 00000000000..9a6b2653605
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/LocalAuthProvisionerSpec.scala
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.web.resource.auth
+
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.{AUTH_PROVIDER, USER}
+import org.apache.texera.dao.jooq.generated.enums.{ProviderTypeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
+import org.jooq.impl.DSL
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import javax.ws.rs.WebApplicationException
+import javax.ws.rs.core.Response
+
+/**
+  * Integration spec for [[LocalAuthProvisioner]]'s unique-violation handling against embedded
+  * Postgres ([[MockTexeraDB]] loads the real `texera_ddl.sql`, so `uq_provider_identity`,
+  * `PRIMARY KEY (uid, provider_type)` and `user_email_key` are the constraints that actually
+  * fire here).
+  *
+  * These paths are only reachable by losing a race, so callers pre-check and the handlers are
+  * the fallback — which is exactly why they need a test rather than a caller. Each case drives
+  * the constraint directly and asserts the message names the cause that actually fired: naming
+  * the wrong one tells a user their free handle is taken, and the two write methods reach
+  * different constraint sets, so a fix applied to one does not cover the other.
+  */
+class LocalAuthProvisionerSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with MockTexeraDB {
+
+  // Shared suffix so cleanup can target this suite's rows precisely; the auth_provider FK is
+  // ON DELETE CASCADE, so deleting the user clears its credential rows.
+  private val emailDomain = "@local-provisioner-test.com"
+
+  private var userDao: UserDao = _
+  private var authDao: AuthProviderDao = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+    authDao = new AuthProviderDao(getDSLContext.configuration())
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  override protected def beforeEach(): Unit = cleanup()
+  override protected def afterEach(): Unit = cleanup()
+
+  private def cleanup(): Unit =
+    getDSLContext.deleteFrom(USER).where(DSL.lower(USER.EMAIL).like("%" + emailDomain)).execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  private def newUser(name: String, localPart: String, placeholder: Boolean = false): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(localPart + emailDomain)
+    user.setRole(UserRoleEnum.INACTIVE)
+    user.setIsPlaceholder(placeholder)
+    user
+  }
+
+  /** Seed a user that already holds a LOCAL credential under `handle`. */
+  private def seedLocalAccount(name: String, localPart: String, handle: String): User = {
+    val user = newUser(name, localPart)
+    user.setRole(UserRoleEnum.REGULAR)
+    userDao.insert(user)
+    val auth = new AuthProvider
+    auth.setUid(user.getUid)
+    auth.setProviderType(ProviderTypeEnum.LOCAL)
+    auth.setProviderId(handle)
+    auth.setPassword(LocalAuthProvisioner.hashPassword("seeded-pw"))
+    authDao.insert(auth)
+    user
+  }
+
+  /** Seed a credential-less placeholder, the row a dataset contributor gets. */
+  private def seedPlaceholder(localPart: String): User = {
+    val user = newUser(localPart, localPart, placeholder = true)
+    userDao.insert(user)
+    user
+  }
+
+  private def userCountByEmail(localPart: String): Int =
+    getDSLContext.fetchCount(USER, DSL.lower(USER.EMAIL).eq(localPart + emailDomain))
+
+  private def conflictFrom(body: => Unit): WebApplicationException = {
+    val thrown = intercept[WebApplicationException](body)
+    thrown.getResponse.getStatus shouldBe Response.Status.CONFLICT.getStatusCode
+    thrown
+  }
+
+  // ---- createLocalAccount --------------------------------------------------
+
+  "LocalAuthProvisioner.createLocalAccount" should
+    "report a taken handle when uq_provider_identity fires" in {
+    seedLocalAccount("Existing", "existing", handle = "shared-handle")
+
+    val thrown = conflictFrom(
+      LocalAuthProvisioner.createLocalAccount(newUser("New", "fresh"), "shared-handle", "pw")
+    )
+
+    thrown.getMessage should include("Login handle shared-handle is already taken")
+    // The user insert and the credential insert share one transaction, so losing the race must
+    // not leave a credential-less account behind under the email that was being registered.
+    userCountByEmail("fresh") shouldBe 0
+  }
+
+  it should "report a registered email when user_email_key fires" in {
+    seedLocalAccount("Existing", "taken-email", handle = "handle-a")
+
+    val thrown = conflictFrom(
+      LocalAuthProvisioner
+        .createLocalAccount(newUser("New", "taken-email"), "handle-b", "pw")
+    )
+
+    thrown.getMessage should include(s"Email taken-email$emailDomain is already registered")
+    // The seeded row is the only one; the losing insert rolled back rather than adding a second.
+    userCountByEmail("taken-email") shouldBe 1
+    getDSLContext.fetchCount(AUTH_PROVIDER, AUTH_PROVIDER.PROVIDER_ID.eq("handle-b")) shouldBe 0
+  }
+
+  // ---- claimWithLocalCredential --------------------------------------------
+
+  "LocalAuthProvisioner.claimWithLocalCredential" should
+    "report an already-claimed account when the (uid, provider_type) primary key fires" in {
+    // A placeholder someone else already claimed: it holds a LOCAL row under *their* handle,
+    // so the handle this caller asks for is free and only the primary key can fire.
+    val claimed = seedPlaceholder("contributor")
+    val auth = new AuthProvider
+    auth.setUid(claimed.getUid)
+    auth.setProviderType(ProviderTypeEnum.LOCAL)
+    auth.setProviderId("first-claimer")
+    auth.setPassword(LocalAuthProvisioner.hashPassword("pw"))
+    authDao.insert(auth)
+
+    claimed.setIsPlaceholder(false)
+    val thrown = conflictFrom(
+      LocalAuthProvisioner.claimWithLocalCredential(claimed, "second-claimer", "pw")
+    )
+
+    thrown.getMessage should include(
+      s"Account for contributor$emailDomain has already been claimed"
+    )
+    // The handle was free, so reporting it as taken would be the wrong cause.
+    thrown.getMessage should not include "already taken"
+    // The claim and the credential share one transaction, so the rollback must leave the
+    // placeholder flag as it was rather than marking an account claimed with no new credential.
+    userDao.fetchOneByUid(claimed.getUid).getIsPlaceholder shouldBe true
+    providerIdOf(claimed.getUid) shouldBe "first-claimer"
+  }
+
+  it should "report a taken handle when uq_provider_identity fires" in {
+    seedLocalAccount("Other", "other", handle = "wanted-handle")
+    val placeholder = seedPlaceholder("claimant")
+
+    placeholder.setIsPlaceholder(false)
+    val thrown = conflictFrom(
+      LocalAuthProvisioner.claimWithLocalCredential(placeholder, "wanted-handle", "pw")
+    )
+
+    thrown.getMessage should include("Login handle wanted-handle is already taken")
+    thrown.getMessage should not include "has already been claimed"
+    userDao.fetchOneByUid(placeholder.getUid).getIsPlaceholder shouldBe true
+  }
+
+  private def providerIdOf(uid: Integer): String =
+    getDSLContext
+      .select(AUTH_PROVIDER.PROVIDER_ID)
+      .from(AUTH_PROVIDER)
+      .where(AUTH_PROVIDER.UID.eq(uid))
+      .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+      .fetchOne(AUTH_PROVIDER.PROVIDER_ID)
+}
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/execution/AdminExecutionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/execution/AdminExecutionResourceSpec.scala
index 16cda7c5a08..89944aff7b2 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/execution/AdminExecutionResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/execution/AdminExecutionResourceSpec.scala
@@ -93,7 +93,6 @@ class AdminExecutionResourceSpec
     testUser.setUid(testUid)
     testUser.setName("test_user")
     testUser.setEmail("admin_exec_test@example.com")
-    testUser.setPassword("password")
 
     testWorkflow = new Workflow
     testWorkflow.setWid(testWid)
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResourceSpec.scala
index fd35653c7a2..b88fe2a8704 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResourceSpec.scala
@@ -21,7 +21,7 @@ package org.apache.texera.web.resource.dashboard.admin.user
 
 import org.apache.texera.dao.MockTexeraDB
 import org.apache.texera.dao.jooq.generated.Tables._
-import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, ProviderTypeEnum, UserRoleEnum}
 import org.apache.texera.dao.jooq.generated.tables.daos.{
   DatasetDao,
   UserDao,
@@ -101,7 +101,6 @@ class AdminUserResourceSpec
       .where(DATASET.OWNER_UID.in(primaryUid, secondaryUid))
       .execute()
     getDSLContext.deleteFrom(USER).where(USER.UID.in(primaryUid, secondaryUid)).execute()
-    // addUser() inserts an INACTIVE user with an auto-generated uid and a "User<millis>" name.
     getDSLContext
       .deleteFrom(USER)
       .where(USER.ROLE.eq(UserRoleEnum.INACTIVE).and(USER.NAME.like("User%")))
@@ -115,11 +114,29 @@ class AdminUserResourceSpec
     user.setEmail(
       s"admin_user_spec_${uid}_${UUID.randomUUID().toString.substring(0, 8)}@example.com"
     )
-    user.setPassword("password")
     user.setRole(role)
     user
   }
 
+  /**
+    * Seed a credential row. `password` is left null for external providers because
+    * ck_provider_credential requires a password for LOCAL and only for LOCAL. The
+    * auth_provider FK is ON DELETE CASCADE, so `cleanup`'s user delete clears these.
+    */
+  private def seedProvider(
+      uid: Int,
+      providerType: ProviderTypeEnum,
+      providerId: String,
+      password: String = null
+  ): Unit =
+    getDSLContext
+      .insertInto(AUTH_PROVIDER)
+      .set(AUTH_PROVIDER.UID, Integer.valueOf(uid))
+      .set(AUTH_PROVIDER.PROVIDER_TYPE, providerType)
+      .set(AUTH_PROVIDER.PROVIDER_ID, providerId)
+      .set(AUTH_PROVIDER.PASSWORD, password)
+      .execute()
+
   private def seedWorkflow(): Workflow = {
     val workflow = new Workflow
     workflow.setWid(testWid)
@@ -191,6 +208,29 @@ class AdminUserResourceSpec
     resource.list().asScala.exists(_.uid == primaryUid) shouldBe false
   }
 
+  // The projection maps onto UserInfo positionally, so a column landing on the wrong field is
+  // silent. Nothing else observes it — pin it here for a user holding both credential kinds,
+  // which also proves the LOCAL row does not leak into the GOOGLE-joined column.
+  it should "report the google id and the avatar for a user with LOCAL and GOOGLE rows" in {
+    val user = makeUser(primaryUid, "dual")
+    user.setAvatar("avatar-blob")
+    userDao.insert(user)
+    seedProvider(primaryUid, ProviderTypeEnum.LOCAL, "dual-handle", password = "hashed")
+    seedProvider(primaryUid, ProviderTypeEnum.GOOGLE, "google-sub-dual")
+
+    val listed = resource.list().asScala.find(_.uid == primaryUid)
+
+    listed.map(u => (u.name, u.googleId, u.googleAvatar)) shouldBe Some(
+      ("dual", "google-sub-dual", "avatar-blob")
+    )
+  }
+
+  it should "leave the google id null for a user with no auth_provider rows" in {
+    userDao.insert(makeUser(primaryUid, "credential-less"))
+
+    resource.list().asScala.find(_.uid == primaryUid).map(_.googleId) shouldBe Some(null)
+  }
+
   // ─── addUser ────────────────────────────────────────────────────────────
 
   "addUser" should "persist a new INACTIVE user" in {
@@ -200,8 +240,17 @@ class AdminUserResourceSpec
 
     val after = userDao.fetchByRole(UserRoleEnum.INACTIVE)
     after.size() shouldBe before + 1
-    // The newly added user has a generated non-empty name and no password left blank.
-    after.asScala.exists(u => u.getName.startsWith("User") && u.getPassword != null) shouldBe true
+    // The newly added user has a generated non-empty name, and the LOCAL credential it logs in
+    // with lives in auth_provider (not on "user"), with its password hash set.
+    after.asScala.exists(u =>
+      u.getName.startsWith("User") && getDSLContext.fetchExists(
+        getDSLContext
+          .selectFrom(AUTH_PROVIDER)
+          .where(AUTH_PROVIDER.UID.eq(u.getUid))
+          .and(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+          .and(AUTH_PROVIDER.PASSWORD.isNotNull)
+      )
+    ) shouldBe true
   }
 
   // ─── updateUser ─────────────────────────────────────────────────────────
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/DatasetResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/DatasetResourceSpec.scala
index 1d4b5635e04..70b8252c096 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/DatasetResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/DatasetResourceSpec.scala
@@ -52,7 +52,6 @@ class DatasetResourceSpec
     user.setName("owner_user")
     user.setRole(UserRoleEnum.ADMIN)
     user.setEmail("owner_user@mail.com")
-    user.setPassword("123")
     user.setComment("test_comment")
     user.setAccountCreationTime(exampleCreationTime)
     user
@@ -64,7 +63,6 @@ class DatasetResourceSpec
     user.setName("test_user")
     user.setEmail("test_user@mail.com")
     user.setRole(UserRoleEnum.REGULAR)
-    user.setPassword("123")
     user.setComment("test_comment2")
     user.setAccountCreationTime(exampleCreationTime)
     user
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala
index 5ec6b88ab2c..d2e0f2bef52 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/file/WorkflowResourceSpec.scala
@@ -71,7 +71,6 @@ class WorkflowResourceSpec
     user.setName("test_user")
     user.setEmail("test_user@mail.com")
     user.setRole(UserRoleEnum.ADMIN)
-    user.setPassword("123")
     user.setComment("test_comment")
     user.setAccountCreationTime(exampleCreationTime)
     user
@@ -83,7 +82,6 @@ class WorkflowResourceSpec
     user.setName("test_user2")
     user.setEmail("test_user2@mail.com")
     user.setRole(UserRoleEnum.ADMIN)
-    user.setPassword("123")
     user.setComment("test_comment2")
     user.setAccountCreationTime(exampleCreationTime)
     user
@@ -221,7 +219,6 @@ class WorkflowResourceSpec
     u.setUid(Integer.valueOf(uid))
     u.setName(s"tmp_user_$uid")
     u.setRole(UserRoleEnum.REGULAR)
-    u.setPassword("pw")
     u.setComment("tmp")
     u.setAccountCreationTime(ts)
     userDao.insert(u)
@@ -272,7 +269,6 @@ class WorkflowResourceSpec
     tmp.setUid(Integer.valueOf(userId))
     tmp.setName("tmp_user")
     tmp.setRole(UserRoleEnum.REGULAR)
-    tmp.setPassword("pw")
     tmp.setComment("tmp")
     // Account creation time not set
     userDao.insert(tmp)
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
index 771250ced88..ffd860541ac 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/hub/HubResourceSpec.scala
@@ -86,7 +86,6 @@ class HubResourceSpec
     u.setUid(Integer.valueOf(uid))
     u.setName(name)
     u.setEmail(s"$name@test.com")
-    u.setPassword("password")
     u
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala
index eecb6ff3cbe..5fd30fee9cc 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala
@@ -84,7 +84,6 @@ class ProjectAccessResourceSpec
     user.setUid(uid)
     user.setName(name)
     user.setEmail(email)
-    user.setPassword("password")
     user.setRole(UserRoleEnum.REGULAR)
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectResourceSpec.scala
index f97d3d45197..7afabeda7f3 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectResourceSpec.scala
@@ -137,7 +137,6 @@ class ProjectResourceSpec
     user.setUid(Integer.valueOf(uid))
     user.setName(name)
     user.setEmail(s"$name@test.com")
-    user.setPassword("password")
     user
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/PublicProjectResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/PublicProjectResourceSpec.scala
index 4d532a9d44e..5f33bc9abee 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/PublicProjectResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/PublicProjectResourceSpec.scala
@@ -53,7 +53,6 @@ class PublicProjectResourceSpec
     user.setEmail(
       s"public_project_spec_${uid}_${UUID.randomUUID().toString.substring(0, 8)}@example.com"
     )
-    user.setPassword("password")
     user.setRole(UserRoleEnum.ADMIN)
     user
   }
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResourceSpec.scala
index 0a0212c2b29..37f4a0a207d 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/quota/UserQuotaResourceSpec.scala
@@ -88,7 +88,6 @@ class UserQuotaResourceSpec extends AnyFlatSpec with BeforeAndAfterAll with Mock
     testUser.setUid(testUid)
     testUser.setName("quota_user")
     testUser.setEmail("quota@example.com")
-    testUser.setPassword("password")
     userDao.insert(testUser)
 
     testWorkflow = new Workflow
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala
index f6ea01c1221..2f6db0c2113 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala
@@ -84,25 +84,21 @@ class WorkflowAccessResourceSpec
     owner.setUid(ownerUid)
     owner.setName("owner")
     owner.setEmail("owner@test.com")
-    owner.setPassword("password")
 
     userWithWrite = new User
     userWithWrite.setUid(userWithWriteUid)
     userWithWrite.setName("user_with_write")
     userWithWrite.setEmail("write@test.com")
-    userWithWrite.setPassword("password")
 
     userWithRead = new User
     userWithRead.setUid(userWithReadUid)
     userWithRead.setName("user_with_read")
     userWithRead.setEmail("read@test.com")
-    userWithRead.setPassword("password")
 
     targetUser = new User
     targetUser.setUid(targetUserUid)
     targetUser.setName("target_user")
     targetUser.setEmail("target@test.com")
-    targetUser.setPassword("password")
 
     // Create test workflow
     testWorkflow = new Workflow
@@ -412,14 +408,14 @@ class WorkflowAccessResourceSpec
     user.setUid(newGranteeUid)
     user.setName(name)
     user.setEmail(email)
-    user.setPassword("password")
     userDao.insert(user)
     user
   }
 
   "WorkflowAccessResource.grantAccess" should "reject granting to a placeholder account" in {
     val placeholder = seedUserWithoutAccess("wf_placeholder", "wf-placeholder@test.com")
-    placeholder.setPassword(null)
+    // a placeholder has no credential at all; under auth_provider that means no row, which
+    // seedUserWithoutAccess already leaves it with
     placeholder.setIsPlaceholder(true)
     userDao.update(placeholder)
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
index 7a3afc03f30..a0a3622e06f 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala
@@ -89,8 +89,7 @@ class WorkflowExecutionsResourceSpec
     testUser.setUid(testUserId)
     testUser.setName("test_user")
     testUser.setEmail("test@example.com")
-    testUser.setPassword("password")
-    testUser.setGoogleAvatar("avatar_url")
+    testUser.setAvatar("avatar_url")
 
     testWorkflow = new Workflow
     testWorkflow.setWid(testWorkflowWid)
@@ -786,7 +785,6 @@ class WorkflowExecutionsResourceSpec
     otherUser.setUid(otherUid)
     otherUser.setName("dataset-owner")
     otherUser.setEmail("owner@example.com")
-    otherUser.setPassword("password")
     userDao.insert(otherUser)
 
     val dataset = new Dataset
@@ -929,7 +927,6 @@ class WorkflowExecutionsResourceSpec
     u.setUid(testUserId + 5000)
     u.setName("no_access_user")
     u.setEmail("noaccess@example.com")
-    u.setPassword("password")
     u
   }
 
@@ -947,6 +944,22 @@ class WorkflowExecutionsResourceSpec
     assert(result.size == 2)
   }
 
+  // fetchInto maps onto WorkflowExecutionEntry POSITIONALLY (a case class has no no-arg
+  // constructor, and jOOQ's mapConstructorParameterNames defaults to false), so `USER.AVATAR`
+  // at projection position 5 lands on `googleAvatar` despite the names differing — exactly as
+  // `last_update_time` at position 9 lands on `completionTime`. Neither mapping was asserted
+  // anywhere before, which is what makes an accidental column reorder silent. Pin both here.
+  it should "map the owner's avatar and completion time onto the entry despite the name mismatch" in {
+    grantReadAccess()
+    insertExecution(lastUpdateOffsetMillis = Some(0L))
+    val entry = resource.retrieveExecutionsOfWorkflow(testWorkflowWid, session(testUser), null).head
+    assert(entry.userName == testUser.getName)
+    assert(entry.googleAvatar == "avatar_url")
+    // `last_update_time` is populated, so a null here would mean position 9 never reached
+    // `completionTime` — i.e. the mapping had silently become name-based.
+    assert(entry.completionTime != null)
+  }
+
   it should "reject an invalid status filter with a BadRequestException" in {
     grantReadAccess()
     assertThrows[BadRequestException](
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResourceCoverSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResourceCoverSpec.scala
index 14ff81eb4d8..3d830dfe09c 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResourceCoverSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResourceCoverSpec.scala
@@ -116,7 +116,6 @@ class WorkflowResourceCoverSpec
     user.setUid(uid)
     user.setName(name)
     user.setEmail(s"$name@test.com")
-    user.setPassword("password")
     user
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
index 06d933f51d0..b633ee83a40 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowVersionResourceSpec.scala
@@ -158,7 +158,6 @@ class WorkflowVersionResourceSpec
     user.setUid(Integer.valueOf(uid))
     user.setName(name)
     user.setEmail(s"$name@test.com")
-    user.setPassword("password")
     user
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
index 9c6e1954fd2..73f9899c831 100644
--- a/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
@@ -110,7 +110,6 @@ class PveResourceSpec
     user.setUid(testUid)
     user.setName("pve_resource_spec_user")
     user.setEmail(s"user_${UUID.randomUUID()}@example.com")
-    user.setPassword("password")
     userDao.insert(user)
   }
 
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
index 74335dd548a..e355c8aef01 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionResultServiceSpec.scala
@@ -93,7 +93,6 @@ class ExecutionResultServiceSpec
     user.setUid(testUid)
     user.setName("execution-result-test-user")
     user.setEmail(s"u$testUid@example.com")
-    user.setPassword("password")
     new UserDao(getDSLContext.configuration()).insert(user)
 
     val workflow = new Workflow
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
index ed67b78cc52..ca704861fe0 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionsMetadataPersistServiceSpec.scala
@@ -88,7 +88,6 @@ class ExecutionsMetadataPersistServiceSpec
     user.setUid(testUid)
     user.setName("metadata_persist_spec_user")
     user.setEmail(s"user_${UUID.randomUUID()}@example.com")
-    user.setPassword("password")
     userDao.insert(user)
 
     val workflow = new Workflow
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
index aa4e5f12c47..9836b1e2262 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ResultExportServiceSpec.scala
@@ -433,7 +433,6 @@ class ResultExportServiceSpec
     testUser.setUid(testUserId)
     testUser.setName("export_user")
     testUser.setEmail("export@example.com")
-    testUser.setPassword("password")
     new UserDao(getDSLContext.configuration()).insert(testUser)
 
     val workflow = new Workflow
diff --git a/common/auth/src/main/scala/org/apache/texera/auth/JwtAuth.scala b/common/auth/src/main/scala/org/apache/texera/auth/JwtAuth.scala
index a97e36a50e4..d3270bfa0a2 100644
--- a/common/auth/src/main/scala/org/apache/texera/auth/JwtAuth.scala
+++ b/common/auth/src/main/scala/org/apache/texera/auth/JwtAuth.scala
@@ -51,14 +51,35 @@ object JwtAuth {
     jws.getCompactSerialization
   }
 
-  def jwtClaims(user: User, expireInDays: Int): JwtClaims = {
+  /**
+    * Build the claim set for `user`. The claim names are a contract with the hand-written
+    * TypeScript reader in `frontend/src/app/common/service/user/auth.service.ts`, which is not
+    * compiled against this file — so renaming one here silently breaks the frontend. `avatar`
+    * now lives on `"user"` rather than a Google-specific column, but the claim keeps its
+    * `googleAvatar` name until the frontend is migrated in lockstep.
+    *
+    * `googleId` is passed in rather than read off `user`, because the GOOGLE provider id lives
+    * in `auth_provider` and this module must stay DB-free: the specs in
+    * `access-control-service` / `config-service` and the token re-issue paths in
+    * `ResultExportService` / `ComputingUnitManagingResource` all call this with no
+    * `auth_provider` context, as does `AuthResource.register`. Omitting the claim is harmless
+    * on all of them: a service-to-service token never reaches the browser, and a freshly
+    * registered LOCAL account has no Google identity to name.
+    *
+    * Note this changes the token's shape, not just where the value comes from: `googleId` was
+    * previously written unconditionally, so a local-only user's token carried `"googleId": null`
+    * and now omits the claim. The frontend declares it optional (`common/type/user.ts`), so the
+    * only reader that can tell is `flarum.service.ts`, which passes it as a Flarum account
+    * password — a path that was already broken for exactly the users who lack the claim.
+    */
+  def jwtClaims(user: User, googleId: Option[String] = None): JwtClaims = {
     val claims = new JwtClaims
     claims.setSubject(user.getName)
     claims.setClaim("userId", user.getUid)
-    claims.setClaim("googleId", user.getGoogleId)
     claims.setClaim("email", user.getEmail)
     claims.setClaim("role", user.getRole)
-    claims.setClaim("googleAvatar", user.getGoogleAvatar)
+    claims.setClaim("googleAvatar", user.getAvatar)
+    googleId.foreach(claims.setClaim("googleId", _))
     claims.setExpirationTimeMinutesInTheFuture(TOKEN_EXPIRE_TIME_IN_MINUTES.toFloat)
     claims
   }
diff --git a/common/auth/src/main/scala/org/apache/texera/auth/JwtParser.scala b/common/auth/src/main/scala/org/apache/texera/auth/JwtParser.scala
index cf324b0959f..6cd540c2851 100644
--- a/common/auth/src/main/scala/org/apache/texera/auth/JwtParser.scala
+++ b/common/auth/src/main/scala/org/apache/texera/auth/JwtParser.scala
@@ -62,8 +62,10 @@ object JwtParser extends LazyLogging {
     // call writes Integer; widen via Number to handle both cases.
     val userId = claims.getClaimValue("userId", classOf[Number]).intValue()
     val role = UserRoleEnum.valueOf(claims.getClaimValue("role").asInstanceOf[String])
-    val googleId = claims.getClaimValue("googleId", classOf[String])
     val googleAvatar = claims.getClaimValue("googleAvatar", classOf[String])
+    // The `googleId` claim is deliberately written but not read back: nothing server-side
+    // needs it (credentials live in auth_provider), and the only consumer is the frontend,
+    // which reads it straight off the raw token.
 
     new SessionUser(
       new User().tap { user =>
@@ -71,8 +73,7 @@ object JwtParser extends LazyLogging {
         user.setName(userName)
         user.setEmail(email)
         user.setRole(role)
-        user.setGoogleId(googleId)
-        user.setGoogleAvatar(googleAvatar)
+        user.setAvatar(googleAvatar)
       }
     )
   }
diff --git a/common/auth/src/main/scala/org/apache/texera/auth/SessionUser.scala b/common/auth/src/main/scala/org/apache/texera/auth/SessionUser.scala
index 709eef1daff..8ac051f8c8f 100644
--- a/common/auth/src/main/scala/org/apache/texera/auth/SessionUser.scala
+++ b/common/auth/src/main/scala/org/apache/texera/auth/SessionUser.scala
@@ -33,7 +33,5 @@ class SessionUser(val user: User) extends Principal {
 
   def getEmail: String = user.getEmail
 
-  def getGoogleId: String = user.getGoogleId
-
   def isRoleOf(role: UserRoleEnum): Boolean = user.getRole == role
 }
diff --git a/common/auth/src/test/scala/org/apache/texera/auth/JwtAuthSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/JwtAuthSpec.scala
index b173ac72128..6ae84a4200d 100644
--- a/common/auth/src/test/scala/org/apache/texera/auth/JwtAuthSpec.scala
+++ b/common/auth/src/test/scala/org/apache/texera/auth/JwtAuthSpec.scala
@@ -33,43 +33,55 @@ class JwtAuthSpec extends AnyFlatSpec with Matchers {
     user.setUid(42)
     user.setName("alice")
     user.setEmail("alice@example.com")
-    user.setGoogleId("g-123")
-    user.setGoogleAvatar("avatar-blob")
+    user.setAvatar("avatar-blob")
     user.setRole(UserRoleEnum.ADMIN)
     user
   }
 
   "JwtAuth.jwtClaims" should "map every User field onto the matching claim" in {
-    val claims = JwtAuth.jwtClaims(buildUser(), 7)
+    val claims = JwtAuth.jwtClaims(buildUser())
     claims.getSubject shouldBe "alice"
     claims.getClaimValueAsString("userId") shouldBe "42"
-    claims.getClaimValueAsString("googleId") shouldBe "g-123"
     claims.getClaimValueAsString("email") shouldBe "alice@example.com"
     claims.getClaimValueAsString("googleAvatar") shouldBe "avatar-blob"
     claims.getClaimValueAsString("role") shouldBe UserRoleEnum.ADMIN.name
   }
 
-  it should "derive the expiration from config, ignoring the expireInDays argument" in {
-    // two very different expireInDays values must yield the same config-derived expiry window
-    def expiryWindowMinutes(expireInDays: Int): Double = {
-      val claims = JwtAuth.jwtClaims(buildUser(), expireInDays)
-      claims.getExpirationTime should not be null
-      claims.getExpirationTime.getValue / 60.0 - NumericDate.now().getValue / 60.0
-    }
-    expiryWindowMinutes(1) shouldBe (AuthConfig.jwtExpirationMinutes.toDouble +- 2.0)
-    expiryWindowMinutes(100000) shouldBe (AuthConfig.jwtExpirationMinutes.toDouble +- 2.0)
+  // Passwords live in auth_provider and never leave it. `googleId` is a different matter: it is
+  // an identifier, not a credential, and the frontend still reads it off the token.
+  it should "not carry any password in the claims" in {
+    val claims = JwtAuth.jwtClaims(buildUser())
+    claims.hasClaim("password") shouldBe false
+    claims.hasClaim("providerId") shouldBe false
+  }
+
+  // The GOOGLE provider id is not on the User pojo any more, so callers with no auth_provider
+  // context (service-to-service token re-issue) simply omit it rather than writing null.
+  it should "omit the googleId claim when no provider id is supplied" in {
+    JwtAuth.jwtClaims(buildUser()).hasClaim("googleId") shouldBe false
+  }
+
+  it should "carry the googleId claim when a provider id is supplied" in {
+    val claims = JwtAuth.jwtClaims(buildUser(), Some("google-sub-123"))
+    claims.getClaimValueAsString("googleId") shouldBe "google-sub-123"
+  }
+
+  it should "derive the expiration from AuthConfig.jwtExpirationMinutes" in {
+    val claims = JwtAuth.jwtClaims(buildUser())
+    claims.getExpirationTime should not be null
+    val windowMinutes = claims.getExpirationTime.getValue / 60.0 - NumericDate.now().getValue / 60.0
+    windowMinutes shouldBe (AuthConfig.jwtExpirationMinutes.toDouble +- 2.0)
   }
 
   it should "produce a token that round-trips back to the same user via JwtParser" in {
-    val token = JwtAuth.jwtToken(JwtAuth.jwtClaims(buildUser(), 1))
+    val token = JwtAuth.jwtToken(JwtAuth.jwtClaims(buildUser()))
     val parsed = JwtParser.parseToken(token)
     parsed.isPresent shouldBe true
     val user = parsed.get().getUser
     user.getUid shouldBe 42
     user.getName shouldBe "alice"
     user.getEmail shouldBe "alice@example.com"
-    user.getGoogleId shouldBe "g-123"
-    user.getGoogleAvatar shouldBe "avatar-blob"
+    user.getAvatar shouldBe "avatar-blob"
     user.getRole shouldBe UserRoleEnum.ADMIN
   }
 
@@ -78,7 +90,7 @@ class JwtAuthSpec extends AnyFlatSpec with Matchers {
     user.setUid(7)
     user.setName("bob")
     user.setRole(UserRoleEnum.ADMIN)
-    val claims = JwtAuth.jwtClaims(user, 1)
+    val claims = JwtAuth.jwtClaims(user)
     claims.getSubject shouldBe "bob"
     claims.getClaimValueAsString("email") shouldBe null
   }
diff --git a/common/auth/src/test/scala/org/apache/texera/auth/JwtParserSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/JwtParserSpec.scala
index dc91de4d645..f18185aad8e 100644
--- a/common/auth/src/test/scala/org/apache/texera/auth/JwtParserSpec.scala
+++ b/common/auth/src/test/scala/org/apache/texera/auth/JwtParserSpec.scala
@@ -38,7 +38,6 @@ class JwtParserSpec extends AnyFlatSpec with Matchers {
     val claims = new JwtClaims
     claims.setSubject("alice")
     claims.setClaim("userId", 42)
-    claims.setClaim("googleId", "g-123")
     claims.setClaim("email", "alice@example.com")
     claims.setClaim("role", UserRoleEnum.ADMIN.name)
     claims.setClaim("googleAvatar", "avatar-blob")
@@ -46,19 +45,17 @@ class JwtParserSpec extends AnyFlatSpec with Matchers {
     claims
   }
 
-  "JwtParser.claimsToSessionUser" should "populate every issued claim including googleAvatar" in {
+  "JwtParser.claimsToSessionUser" should "populate every issued claim including avatar" in {
     val user: User = JwtParser.claimsToSessionUser(buildClaims()).getUser
     user.getUid shouldBe 42
     user.getName shouldBe "alice"
     user.getEmail shouldBe "alice@example.com"
-    user.getGoogleId shouldBe "g-123"
-    user.getGoogleAvatar shouldBe "avatar-blob"
+    user.getAvatar shouldBe "avatar-blob"
     user.getRole shouldBe UserRoleEnum.ADMIN
   }
 
-  it should "leave non-issued slots null (password, comment, accountCreation, affiliation, joiningReason)" in {
+  it should "leave non-issued slots null (comment, accountCreation, affiliation, joiningReason)" in {
     val user: User = JwtParser.claimsToSessionUser(buildClaims()).getUser
-    user.getPassword shouldBe null
     user.getComment shouldBe null
     user.getAccountCreationTime shouldBe null
     user.getAffiliation shouldBe null
@@ -71,7 +68,7 @@ class JwtParserSpec extends AnyFlatSpec with Matchers {
     parsed.isPresent shouldBe true
     val u = parsed.get().getUser
     u.getUid shouldBe 42
-    u.getGoogleAvatar shouldBe "avatar-blob"
+    u.getAvatar shouldBe "avatar-blob"
   }
 
   "JwtParser.parseToken" should "return empty on a structurally invalid token" in {
@@ -159,7 +156,6 @@ class JwtParserSpec extends AnyFlatSpec with Matchers {
     val bob = new JwtClaims
     bob.setSubject("bob")
     bob.setClaim("userId", 7)
-    bob.setClaim("googleId", "g-bob")
     bob.setClaim("email", "bob@example.com")
     bob.setClaim("role", UserRoleEnum.REGULAR.name)
     bob.setClaim("googleAvatar", "bob-avatar")
@@ -183,7 +179,7 @@ class JwtParserSpec extends AnyFlatSpec with Matchers {
     first.getUid shouldBe second.getUid
     first.getName shouldBe second.getName
     first.getEmail shouldBe second.getEmail
-    first.getGoogleAvatar shouldBe second.getGoogleAvatar
+    first.getAvatar shouldBe second.getAvatar
     first.getRole shouldBe second.getRole
   }
 
diff --git a/common/auth/src/test/scala/org/apache/texera/auth/SessionUserSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/SessionUserSpec.scala
index 4dc7682a236..5a7c2cbeebd 100644
--- a/common/auth/src/test/scala/org/apache/texera/auth/SessionUserSpec.scala
+++ b/common/auth/src/test/scala/org/apache/texera/auth/SessionUserSpec.scala
@@ -31,7 +31,6 @@ class SessionUserSpec extends AnyFlatSpec with Matchers {
     user.setUid(42)
     user.setName("alice")
     user.setEmail("alice@example.com")
-    user.setGoogleId("g-123")
     user.setRole(role)
     user
   }
@@ -54,12 +53,6 @@ class SessionUserSpec extends AnyFlatSpec with Matchers {
     session.getEmail shouldBe user.getEmail
   }
 
-  it should "expose the underlying User's googleId via getGoogleId" in {
-    val user = buildUser()
-    val session = new SessionUser(user)
-    session.getGoogleId shouldBe user.getGoogleId
-  }
-
   it should "return the same User instance via getUser" in {
     val user = buildUser()
     val session = new SessionUser(user)
diff --git a/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala b/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala
index d6589ba4d05..479d378f632 100644
--- a/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala
+++ b/common/auth/src/test/scala/org/apache/texera/auth/util/ComputingUnitAccessSpec.scala
@@ -51,7 +51,6 @@ class ComputingUnitAccessSpec
     val u = new User
     u.setUid(uid)
     u.setName(name)
-    u.setPassword("password")
     u
   }
 
diff --git a/common/dao/src/main/scala/org/apache/texera/dao/SqlStates.scala b/common/dao/src/main/scala/org/apache/texera/dao/SqlStates.scala
new file mode 100644
index 00000000000..0f9d76bffb9
--- /dev/null
+++ b/common/dao/src/main/scala/org/apache/texera/dao/SqlStates.scala
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.dao
+
+/**
+  * SQLSTATE codes callers match on when turning a `DataAccessException` into an HTTP response.
+  *
+  * These live beside [[SqlServer]] rather than in whichever resource happens to need one first:
+  * the code is a property of the database, not of any one endpoint, and the callers that catch it
+  * sit in unrelated packages. A constant owned by one of those packages is unreachable from the
+  * rest, so each would keep its own literal.
+  */
+object SqlStates {
+
+  /** Postgres unique-violation: a unique constraint or primary key was already satisfied. */
+  val UNIQUE_VIOLATION = "23505"
+}
diff --git a/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala b/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala
index 23c549d0595..30c09b152ff 100644
--- a/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala
+++ b/common/dao/src/test/scala/org/apache/texera/dao/UserWarehouseSpec.scala
@@ -44,10 +44,12 @@ class UserWarehouseSpec extends AnyFlatSpec with Matchers with BeforeAndAfterAll
     try closeConnectionPool()
     finally super.afterAll()
 
+  // Credentials live in auth_provider now, and the ck_nulltest constraint that used to require
+  // one on the user row went with them, so a warehouse owner needs nothing but a name.
   private def insertUser(name: String): Integer =
     getDSLContext
-      .insertInto(USER, USER.NAME, USER.PASSWORD)
-      .values(name, "password")
+      .insertInto(USER, USER.NAME)
+      .values(name)
       .returning(USER.UID)
       .fetchOne()
       .getUid
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
index 6916ec66410..7056ee5304f 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
@@ -41,7 +41,6 @@ class FileResolverSpec
     user.setUid(Integer.valueOf(1))
     user.setName("test_user")
     user.setRole(UserRoleEnum.ADMIN)
-    user.setPassword("123")
     user.setEmail("test_user@test.com")
     user
   }
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
index 778e5ed02a5..6e8a165e3ba 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
@@ -26,7 +26,7 @@ import jakarta.annotation.security.RolesAllowed
 import jakarta.ws.rs._
 import jakarta.ws.rs.core.{MediaType, Response}
 import org.apache.commons.lang3.StringUtils
-import org.apache.texera.auth.JwtAuth.{TOKEN_EXPIRE_TIME_IN_MINUTES, jwtClaims}
+import org.apache.texera.auth.JwtAuth.jwtClaims
 import org.apache.texera.auth.{JwtAuth, SessionUser}
 import org.apache.texera.common.config.KubernetesConfig.{
   cpuLimitOptions,
@@ -380,7 +380,7 @@ class ComputingUnitManagingResource {
       }
 
       val computingUnit = new WorkflowComputingUnit()
-      val userToken = JwtAuth.jwtToken(jwtClaims(user.user, TOKEN_EXPIRE_TIME_IN_MINUTES))
+      val userToken = JwtAuth.jwtToken(jwtClaims(user.user))
       computingUnit.setUid(user.getUid)
       computingUnit.setName(param.name)
       computingUnit.setCreationTime(new Timestamp(System.currentTimeMillis()))
@@ -399,7 +399,7 @@ class ComputingUnitManagingResource {
       val userDao = new UserDao(ctx.configuration())
       val ownerUser = Option(userDao.fetchOneByUid(user.getUid))
       val ownerGoogleAvatar: String =
-        ownerUser.flatMap(u => Option(u.getGoogleAvatar).filter(_.nonEmpty)).orNull
+        ownerUser.flatMap(u => Option(u.getAvatar).filter(_.nonEmpty)).orNull
       val ownerUsername: String =
         ownerUser.flatMap(u => Option(u.getName).filter(_.nonEmpty)).orNull
 
@@ -559,7 +559,7 @@ class ComputingUnitManagingResource {
     val userDao = new UserDao(context.configuration())
     val ownerUser = Option(userDao.fetchOneByUid(unit.getUid))
     val ownerGoogleAvatar: String =
-      ownerUser.flatMap(u => Option(u.getGoogleAvatar).filter(_.nonEmpty)).orNull
+      ownerUser.flatMap(u => Option(u.getAvatar).filter(_.nonEmpty)).orNull
     val ownerUsername: String =
       ownerUser.flatMap(u => Option(u.getName).filter(_.nonEmpty)).orNull
 
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
index ea7a65ef517..aaff9943d9a 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/ComputingUnitHelpers.scala
@@ -47,7 +47,7 @@ object ComputingUnitHelpers {
         .fetchByUid(uids: _*)
         .asScala
         .map { u =>
-          val avatar = Option(u.getGoogleAvatar).filter(_.nonEmpty).orNull
+          val avatar = Option(u.getAvatar).filter(_.nonEmpty).orNull
           val name = Option(u.getName).filter(_.nonEmpty).orNull
           u.getUid -> (avatar, name)
         }
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
index a721986b695..f3d2f9e498e 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/AdminComputingUnitResourceSpec.scala
@@ -67,8 +67,10 @@ class AdminComputingUnitResourceSpec
     u.setName(name)
     u.setEmail(s"user$uid@example.com")
     u.setRole(UserRoleEnum.ADMIN)
-    u.setPassword("password")
-    u.setGoogleAvatar(s"avatar-$uid")
+    // Credentials live in auth_provider now, and this spec exercises the listing rather than
+    // login, so the user needs none. The avatar column is provider-neutral; the DTO field it
+    // feeds is still named ownerGoogleAvatar.
+    u.setAvatar(s"avatar-$uid")
     u
   }
 
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
index 94a97dc6c2b..1396784f372 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessResourceSpec.scala
@@ -65,7 +65,6 @@ class ComputingUnitAccessResourceSpec
   private val ownerUser: User = {
     val user = new User
     user.setName("cu_owner")
-    user.setPassword("123")
     user.setEmail("cu_owner@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -74,7 +73,6 @@ class ComputingUnitAccessResourceSpec
   private val granteeUser: User = {
     val user = new User
     user.setName("cu_grantee")
-    user.setPassword("123")
     user.setEmail("cu_grantee@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -83,7 +81,6 @@ class ComputingUnitAccessResourceSpec
   private val strangerUser: User = {
     val user = new User
     user.setName("cu_stranger")
-    user.setPassword("123")
     user.setEmail("cu_stranger@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessSharingDisabledSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessSharingDisabledSpec.scala
index 675ad5ccddd..ffd92658fa8 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessSharingDisabledSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitAccessSharingDisabledSpec.scala
@@ -48,7 +48,6 @@ class ComputingUnitAccessSharingDisabledSpec
   private val user: User = {
     val u = new User
     u.setName("cu_user")
-    u.setPassword("123")
     u.setEmail("cu_user@test.com")
     u.setRole(UserRoleEnum.REGULAR)
     u
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
index 872b5e23b6d..e1142a63466 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
@@ -49,8 +49,7 @@ class ComputingUnitManagingResourceSpec
     u.setName("owner")
     u.setEmail("owner@example.com")
     u.setRole(UserRoleEnum.REGULAR)
-    u.setPassword("password")
-    u.setGoogleAvatar("owner-avatar")
+    u.setAvatar("owner-avatar")
     new SessionUser(u)
   }
 
@@ -62,7 +61,8 @@ class ComputingUnitManagingResourceSpec
     u.setName(name)
     u.setEmail(s"$name@example.com")
     u.setRole(role)
-    u.setPassword("password")
+    // Credentials live in auth_provider now, and this spec exercises computing-unit ownership
+    // rather than login, so the user needs none.
     u
   }
   private lazy val adminUser: SessionUser =
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
index cf54089a132..05e6fe88bbc 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/ComputingUnitHelpersSpec.scala
@@ -59,8 +59,7 @@ class ComputingUnitHelpersSpec
     u.setName(name)
     u.setEmail(email)
     u.setRole(UserRoleEnum.REGULAR)
-    u.setPassword("password")
-    u.setGoogleAvatar(avatar)
+    u.setAvatar(avatar)
     u
   }
 
diff --git a/config-service/src/test/scala/org/apache/texera/service/resource/ConfigResourceSpec.scala b/config-service/src/test/scala/org/apache/texera/service/resource/ConfigResourceSpec.scala
index f27f9463d8e..295de41c563 100644
--- a/config-service/src/test/scala/org/apache/texera/service/resource/ConfigResourceSpec.scala
+++ b/config-service/src/test/scala/org/apache/texera/service/resource/ConfigResourceSpec.scala
@@ -103,9 +103,8 @@ class ConfigResourceSpec
     u.setUid(2)
     u.setName("test-regular")
     u.setEmail("test-regular@example.com")
-    u.setGoogleId(null)
     u.setRole(UserRoleEnum.REGULAR)
-    JwtAuth.jwtToken(JwtAuth.jwtClaims(u, expireInDays = 1))
+    JwtAuth.jwtToken(JwtAuth.jwtClaims(u))
   }
 
   private def adminToken(): String = {
@@ -113,9 +112,8 @@ class ConfigResourceSpec
     u.setUid(1)
     u.setName("test-admin")
     u.setEmail("test-admin@example.com")
-    u.setGoogleId(null)
     u.setRole(UserRoleEnum.ADMIN)
-    JwtAuth.jwtToken(JwtAuth.jwtClaims(u, expireInDays = 1))
+    JwtAuth.jwtToken(JwtAuth.jwtClaims(u))
   }
 
   "GET /config/pre-login" should "return 200 without an Authorization header" in {
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
index 1fc7b2e2e79..f73c7adf642 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala
@@ -50,7 +50,6 @@ class DatasetAccessResourceSpec
   private val ownerUser: User = {
     val user = new User
     user.setName("dataset_owner")
-    user.setPassword("123")
     user.setEmail("dataset_owner@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -59,7 +58,6 @@ class DatasetAccessResourceSpec
   private val readGranteeUser: User = {
     val user = new User
     user.setName("read_grantee")
-    user.setPassword("123")
     user.setEmail("read_grantee@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -68,7 +66,6 @@ class DatasetAccessResourceSpec
   private val writeGranteeUser: User = {
     val user = new User
     user.setName("write_grantee")
-    user.setPassword("123")
     user.setEmail("write_grantee@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -77,7 +74,6 @@ class DatasetAccessResourceSpec
   private val strangerUser: User = {
     val user = new User
     user.setName("stranger")
-    user.setPassword("123")
     user.setEmail("stranger@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
index e9e673280a3..8d89ffe578c 100644
--- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
@@ -27,6 +27,7 @@ import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
 import org.apache.texera.auth.SessionUser
 import org.apache.texera.dao.MockTexeraDB
 import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum}
+import org.apache.texera.dao.jooq.generated.tables.AuthProvider.AUTH_PROVIDER
 import org.apache.texera.dao.jooq.generated.tables.DatasetUploadSession.DATASET_UPLOAD_SESSION
 import org.apache.texera.dao.jooq.generated.tables.DatasetUploadSessionPart.DATASET_UPLOAD_SESSION_PART
 import org.apache.texera.dao.jooq.generated.tables.daos.{
@@ -137,7 +138,6 @@ class DatasetResourceSpec
   private val ownerUser: User = {
     val user = new User
     user.setName("test_user")
-    user.setPassword("123")
     user.setEmail("test_user@test.com")
     user.setRole(UserRoleEnum.ADMIN)
     user
@@ -146,7 +146,6 @@ class DatasetResourceSpec
   private val otherAdminUser: User = {
     val user = new User
     user.setName("test_user2")
-    user.setPassword("123")
     user.setEmail("test_user2@test.com")
     user.setRole(UserRoleEnum.ADMIN)
     user
@@ -156,7 +155,6 @@ class DatasetResourceSpec
   private val multipartNoWriteUser: User = {
     val user = new User
     user.setName("multipart_user2")
-    user.setPassword("123")
     user.setEmail("multipart_user2@test.com")
     user.setRole(UserRoleEnum.REGULAR)
     user
@@ -674,7 +672,11 @@ class DatasetResourceSpec
     placeholder should not be null
     placeholder.getIsPlaceholder shouldBe true
     placeholder.getRole shouldEqual UserRoleEnum.INACTIVE
-    placeholder.getPassword shouldBe null
+    // credentials live in auth_provider now, and a placeholder has none at all
+    getDSLContext.fetchCount(
+      AUTH_PROVIDER,
+      AUTH_PROVIDER.UID.eq(placeholder.getUid)
+    ) shouldBe 0
 
     val firstUid = DatasetResource.getContributorsByDid(getDSLContext, did).head.uid
 
diff --git a/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala b/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala
index 2445af02f77..1a5f9bd0c75 100644
--- a/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/util/StagedFileCleanupJobSpec.scala
@@ -86,7 +86,6 @@ class StagedFileCleanupJobSpec
   private val ownerUser: User = {
     val user = new User
     user.setName("cleanup_test_user")
-    user.setPassword("123")
     user.setEmail("cleanup_test_user@test.com")
     user.setRole(UserRoleEnum.ADMIN)
     user
diff --git a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
index ea6505a86b7..bb1a29c7e57 100644
--- a/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
+++ b/notebook-migration-service/src/test/scala/org/apache/texera/service/resource/NotebookMigrationResourceSpec.scala
@@ -120,7 +120,6 @@ class NotebookMigrationResourceSpec
     user.setName(name)
     user.setEmail(email)
     user.setRole(UserRoleEnum.REGULAR)
-    user.setPassword("password")
     userDao.insert(user)
     user.getUid
   }
diff --git a/sql/changelog.xml b/sql/changelog.xml
index f6daf9dc33c..586618f39b7 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -73,10 +73,16 @@
         <sqlFile path="sql/updates/32.sql"/>
     </changeSet>
 
+    <!-- Split auth credentials out of "user" into auth_provider, and rename the
+         Google-specific google_avatar column to the provider-neutral avatar -->
+    <changeSet id="33" author="Neilk1021">
+        <sqlFile path="sql/updates/33.sql"/>
+    </changeSet>
+
     <!-- example changeSet
     <changeSet id="1" author="author">
         <sqlFile path="sql/updates/1.sql"/>
     </changeSet>
     -->
 
-</databaseChangeLog>
\ No newline at end of file
+</databaseChangeLog>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index e90b79226a2..45ade5b9109 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -52,6 +52,7 @@ DROP TABLE IF EXISTS operator_port_cache CASCADE;
 DROP TABLE IF EXISTS workflow_user_access CASCADE;
 DROP TABLE IF EXISTS workflow_of_user CASCADE;
 DROP TABLE IF EXISTS user_config CASCADE;
+DROP TABLE IF EXISTS auth_provider CASCADE;
 DROP TABLE IF EXISTS "user" CASCADE;
 DROP TABLE IF EXISTS user_last_active_time CASCADE;
 DROP TABLE IF EXISTS workflow CASCADE;
@@ -87,11 +88,13 @@ DROP TABLE IF EXISTS virtual_environments CASCADE;
 DROP TYPE IF EXISTS user_role_enum CASCADE;
 DROP TYPE IF EXISTS privilege_enum CASCADE;
 DROP TYPE IF EXISTS action_enum CASCADE;
+DROP TYPE IF EXISTS provider_type_enum CASCADE;
 
 CREATE TYPE user_role_enum AS ENUM ('INACTIVE', 'RESTRICTED', 'REGULAR', 'ADMIN');
 CREATE TYPE action_enum AS ENUM ('like', 'unlike', 'view', 'clone');
 CREATE TYPE privilege_enum AS ENUM ('NONE', 'READ', 'WRITE');
 CREATE TYPE workflow_computing_unit_type_enum AS ENUM ('local', 'kubernetes');
+CREATE TYPE provider_type_enum AS ENUM ('LOCAL', 'GOOGLE');
 CREATE TYPE user_warehouse_flavor_enum AS ENUM ('local', 'aws');
 
 -- ============================================
@@ -104,18 +107,27 @@ CREATE TABLE IF NOT EXISTS "user"
     uid                     SERIAL PRIMARY KEY,
     name                    VARCHAR(256) NOT NULL,
     email                   VARCHAR(256) UNIQUE,
-    password                VARCHAR(256),
-    google_id               VARCHAR(256) UNIQUE,
-    google_avatar           VARCHAR(100),
+    avatar                  VARCHAR(100),
     role                    user_role_enum NOT NULL DEFAULT 'INACTIVE',
     comment                 TEXT,
     account_creation_time   TIMESTAMPTZ NOT NULL DEFAULT now(),
     affiliation             VARCHAR(128),
     joining_reason          VARCHAR(500),
     -- placeholder accounts are auto-created for dataset contributors and carry no credentials until claimed
-    is_placeholder          BOOLEAN NOT NULL DEFAULT FALSE,
-    -- every non-placeholder account must have a credential
-    CONSTRAINT ck_nulltest CHECK ((password IS NOT NULL) OR (google_id IS NOT NULL) OR is_placeholder)
+    is_placeholder          BOOLEAN NOT NULL DEFAULT FALSE
+    );
+
+CREATE TABLE IF NOT EXISTS auth_provider
+(
+    uid               INT                 NOT NULL,
+    provider_type     provider_type_enum  NOT NULL,
+    provider_id       VARCHAR(256)        NOT NULL,
+    password          VARCHAR(256), -- hashed credential; only for LOCAL
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (uid, provider_type),
+    FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE,
+    CONSTRAINT uq_provider_identity UNIQUE (provider_type, provider_id),
+    CONSTRAINT ck_provider_credential CHECK ((provider_type = 'LOCAL') = (password IS NOT NULL))
     );
 
 -- Contributor emails are resolved with lower(email) lookups.
diff --git a/sql/updates/33.sql b/sql/updates/33.sql
new file mode 100644
index 00000000000..7f1a821de0e
--- /dev/null
+++ b/sql/updates/33.sql
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+-- Relocate login credentials out of "user" into auth_provider.
+--
+-- Moves `password` / `google_id` into an auth_provider row per (user, provider), so a user can
+-- hold several external identities instead of exactly one Google account, and renames
+-- `google_avatar` to the provider-neutral `avatar`. The rename is in place: the column keeps
+-- its width and every stored value, so this migration does not change what any user's avatar
+-- resolves to.
+
+\c texera_db
+
+SET search_path TO texera_db;
+
+BEGIN;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'provider_type_enum') THEN
+        CREATE TYPE provider_type_enum AS ENUM ('LOCAL', 'GOOGLE');
+    END IF;
+END
+$$;
+
+-- provider_id is nullable here and tightened to NOT NULL below, once the backfill has given
+-- every row a handle.
+CREATE TABLE IF NOT EXISTS auth_provider
+(
+    uid               INT                 NOT NULL,
+    provider_type     provider_type_enum  NOT NULL,
+    provider_id       VARCHAR(256),
+    password          VARCHAR(256), -- hashed credential; only for LOCAL
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (uid, provider_type),
+    FOREIGN KEY (uid) REFERENCES "user"(uid) ON DELETE CASCADE
+);
+
+-- Both constraints are (re-)added at the end, so drop them first to keep this file re-runnable
+-- on a database that already has them.
+ALTER TABLE auth_provider DROP CONSTRAINT IF EXISTS ck_provider_credential;
+ALTER TABLE auth_provider DROP CONSTRAINT IF EXISTS uq_provider_identity;
+
+-- Report the accounts that end up unable to log in. This only reports: a name that is blank,
+-- padded or shared is normalized into a usable handle further down rather than rejected, because
+-- refusing the migration over a cosmetic name turns one untrimmed row — which `AdminUserResource`
+-- can write today — into a deployment that cannot start, and liquibase marks the changeset failed.
+DO $$
+DECLARE
+    orphans   TEXT;
+    has_placeholder BOOLEAN;
+BEGIN
+    -- `is_placeholder` accounts (migration 31) deliberately have no credential, so they are
+    -- not orphans and must not be reported as such. Checked dynamically because this migration
+    -- also has to run against databases predating that column.
+    SELECT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'is_placeholder'
+    ) INTO has_placeholder;
+
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'password'
+    ) THEN
+        IF has_placeholder THEN
+            EXECUTE $q$
+                SELECT string_agg(uid::TEXT, ', ')
+                FROM "user"
+                WHERE password IS NULL AND google_id IS NULL AND NOT is_placeholder
+            $q$ INTO orphans;
+        ELSE
+            SELECT string_agg(uid::TEXT, ', ')
+            INTO orphans
+            FROM "user"
+            WHERE password IS NULL AND google_id IS NULL;
+        END IF;
+    END IF;
+
+    IF orphans IS NOT NULL THEN
+        RAISE NOTICE 'migration 33: uid(s) % have neither a password nor a google_id, so they '
+                     'get no auth_provider row and cannot log in.', orphans;
+    END IF;
+END
+$$;
+
+-- Backfill one auth_provider row per credential the user already had.
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'password'
+    ) THEN
+        -- The handle is left NULL here and minted below, where trimming, blank names and
+        -- collisions are all resolved in one place; `provider_id IS NULL` is what marks a row
+        -- as not yet minted.
+        INSERT INTO auth_provider (uid, provider_type, password)
+        SELECT uid, 'LOCAL'::provider_type_enum, password
+        FROM "user"
+        WHERE password IS NOT NULL
+        ON CONFLICT (uid, provider_type) DO NOTHING;
+
+        INSERT INTO auth_provider (uid, provider_type, provider_id)
+        SELECT uid, 'GOOGLE'::provider_type_enum, google_id
+        FROM "user"
+        WHERE google_id IS NOT NULL
+        ON CONFLICT (uid, provider_type) DO NOTHING;
+    END IF;
+END
+$$;
+
+-- Mint every LOCAL handle from "user".name, normalizing rather than rejecting: the name is
+-- trimmed, a name with nothing left is replaced by a uid-derived handle, and handles that still
+-- collide are deterministically suffixed with "-<uid>" (kept: the lowest uid), truncated to fit
+-- VARCHAR(256). The loop is bounded because a suffixed handle can itself collide with a literal
+-- one, the same way 28.sql deduplicates dataset names.
+--
+-- Every change is reported via RAISE NOTICE: the handle is what the user types to log in, so an
+-- operator has to be able to see which accounts got one they would not guess and tell them.
+DO $$
+DECLARE
+    rec RECORD;
+    changed    INT := 0;
+    iterations INT := 0;
+BEGIN
+    -- Trim, give a name that is blank or whitespace-only a deterministic stand-in, and separate
+    -- names that are already shared. Separating them here rather than leaving all of it to the
+    -- loop below settles the common case in one pass, so the loop only has to resolve the
+    -- residue: a minted "-<uid>" handle that collides with a literal one.
+    FOR rec IN
+        UPDATE auth_provider a
+        SET provider_id = CASE
+                              WHEN src.rn = 1 THEN src.base
+                              ELSE LEFT(src.base, 256 - LENGTH('-' || a.uid::TEXT))
+                                       || '-' || a.uid::TEXT
+                          END
+        FROM (
+            SELECT u.uid,
+                   u.name AS old_name,
+                   CASE
+                       WHEN btrim(u.name) = '' THEN 'user-' || u.uid::TEXT
+                       ELSE btrim(u.name)
+                   END AS base,
+                   ROW_NUMBER() OVER (
+                       PARTITION BY CASE
+                                        WHEN btrim(u.name) = '' THEN 'user-' || u.uid::TEXT
+                                        ELSE btrim(u.name)
+                                    END
+                       ORDER BY u.uid
+                   ) AS rn
+            FROM "user" u
+                     JOIN auth_provider ap
+                          ON ap.uid = u.uid
+                              AND ap.provider_type = 'LOCAL'
+                              AND ap.provider_id IS NULL
+        ) src
+        WHERE a.uid = src.uid
+          AND a.provider_type = 'LOCAL'
+          AND a.provider_id IS NULL
+        RETURNING a.uid, src.old_name, a.provider_id AS new_handle
+    LOOP
+        IF rec.old_name IS DISTINCT FROM rec.new_handle THEN
+            changed := changed + 1;
+            RAISE NOTICE 'migration 33: minted LOCAL handle for uid=%: "%" -> "%"',
+                rec.uid, rec.old_name, rec.new_handle;
+        END IF;
+    END LOOP;
+
+    -- Resolve handles shared by more than one account.
+    LOOP
+        FOR rec IN
+            UPDATE auth_provider a
+            SET provider_id = LEFT(a.provider_id, 256 - LENGTH('-' || a.uid::TEXT))
+                                  || '-' || a.uid::TEXT
+            FROM (
+                SELECT uid, provider_id AS old_handle,
+                       ROW_NUMBER() OVER (PARTITION BY provider_id ORDER BY uid) AS rn
+                FROM auth_provider
+                WHERE provider_type = 'LOCAL'
+            ) dups
+            WHERE a.uid = dups.uid AND a.provider_type = 'LOCAL' AND dups.rn > 1
+            RETURNING a.uid, dups.old_handle, a.provider_id AS new_handle
+        LOOP
+            changed := changed + 1;
+            RAISE NOTICE 'migration 33: LOCAL handle for uid=% was already taken: "%" -> "%"',
+                rec.uid, rec.old_handle, rec.new_handle;
+        END LOOP;
+
+        EXIT WHEN NOT EXISTS (
+            SELECT 1 FROM auth_provider
+            WHERE provider_type = 'LOCAL'
+            GROUP BY provider_id HAVING COUNT(*) > 1
+        );
+
+        iterations := iterations + 1;
+        IF iterations > 10 THEN
+            RAISE EXCEPTION 'migration 33: could not make LOCAL login handles unique after 10 '
+                            'passes; resolve the duplicates in "user".name manually and re-run.';
+        END IF;
+    END LOOP;
+
+    IF changed > 0 THEN
+        RAISE NOTICE 'migration 33: % LOCAL login handle(s) differ from the account name they '
+                     'were minted from; those users cannot guess their handle and must be told '
+                     'it or given a reset.', changed;
+    END IF;
+END
+$$;
+
+-- Every row now has a handle, so make it mandatory, enforce uniqueness, and restore the
+-- credential check in its new shape: a password exists for LOCAL and only for LOCAL.
+--
+-- uq_provider_identity belongs here rather than in CREATE TABLE: Postgres checks a
+-- non-deferrable UNIQUE as each index tuple is inserted, so in force it aborts the minting
+-- UPDATE above the moment that derives a colliding handle, before the loop can resolve it.
+-- Same ordering as 28.sql, which deduplicates before adding dataset_owner_uid_name_key.
+ALTER TABLE auth_provider ALTER COLUMN provider_id SET NOT NULL;
+ALTER TABLE auth_provider
+    ADD CONSTRAINT uq_provider_identity UNIQUE (provider_type, provider_id);
+ALTER TABLE auth_provider
+    ADD CONSTRAINT ck_provider_credential CHECK ((provider_type = 'LOCAL') = (password IS NOT NULL));
+
+-- Keep the avatar as a provider-neutral profile column on "user" (rename in place).
+-- Guarded so it is a no-op on a fresh DB where "user" already has "avatar".
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'google_avatar'
+    ) AND NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'avatar'
+    ) THEN
+        ALTER TABLE "user" RENAME COLUMN google_avatar TO avatar;
+    END IF;
+END
+$$;
+
+-- ck_nulltest constrained password/google_id, which are about to disappear. Its "every user
+-- has a credential" rule cannot be a row-level check once credentials live in a child table,
+-- so it is dropped rather than reshaped: a user with no auth_provider row is now legal and
+-- simply cannot log in.
+ALTER TABLE "user" DROP CONSTRAINT IF EXISTS ck_nulltest;
+ALTER TABLE "user" DROP COLUMN IF EXISTS password;
+ALTER TABLE "user" DROP COLUMN IF EXISTS google_id;
+
+COMMIT;