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..b2214eb2b97 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
   }
 
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..77c31c62860 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,7 +164,6 @@ 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))
   }
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 277180d2d52..dd5c4dce6f7 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,13 +19,14 @@
 
 package org.apache.texera.web.resource.auth
 
+import com.typesafe.scalalogging.Logger
 import org.apache.texera.auth.JwtAuth.{TOKEN_EXPIRE_TIME_IN_MINUTES, jwtClaims, jwtToken}
 import org.apache.texera.common.config.UserSystemConfig
 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.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.apache.texera.web.model.http.response.TokenIssueResponse
 import org.apache.texera.web.resource.auth.AuthResource._
@@ -35,53 +36,103 @@ 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
+  private def userDao = new UserDao(context.configuration)
+
+  private val passwordEncryptor = new StrongPasswordEncryptor
+
+  private def localHandleExists(handle: String): Boolean = {
+    context.fetchExists(
+      context
+        .selectFrom(AUTH_PROVIDER)
+        .where(AUTH_PROVIDER.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+        .and(AUTH_PROVIDER.PROVIDER_ID.eq(handle))
     )
+  }
+
+  //TODO ASSERT THAT ALL USERS WERE MIGRATED CORRECTLY AND CHECK
 
   /**
     * 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 String
     * @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
 
-  def createAdminUser(): Unit = {
-    val adminUsername = UserSystemConfig.adminUsername
-    val adminPassword = UserSystemConfig.adminPassword
+    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()
 
-    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)
+    Option(record).flatMap(r => {
+      val encryptedPassword = r.get(AUTH_PROVIDER.PASSWORD)
+      if (passwordEncryptor.checkPassword(password, encryptedPassword)) {
+        Some(r.into(USER).into(classOf[User]))
+      } else {
+        None
       }
+    })
+  }
+
+  /**
+    * Create a user together with the LOCAL credential it logs in with. The handle is passed
+    * explicitly rather than read off `user.getName`, so that identity is never re-derived
+    * from the mutable display name.
+    */
+  private def insertLocalUser(user: User, handle: String, hashedPassword: String): Unit = {
+    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)
     }
   }
+
+  def createAdminUser(): Unit =
+    createAdminUser(UserSystemConfig.adminUsername.trim, UserSystemConfig.adminPassword.trim)
+
+  /**
+    * 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 (localHandleExists(adminUsername)) return
+
+    if (userDao.fetchOneByEmail(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)
+
+    insertLocalUser(user, adminUsername, passwordEncryptor.encryptPassword(adminPassword))
+  }
 }
 
 @Path("/auth/")
@@ -128,9 +179,11 @@ class AuthResource {
         user.setName(username)
         user.setEmail(useremail)
         user.setRole(UserRoleEnum.RESTRICTED)
-        // hash the plain text password
-        user.setPassword(new StrongPasswordEncryptor().encryptPassword(userpassword))
-        userDao.insert(user)
+        insertLocalUser(
+          user,
+          username,
+          AuthResource.passwordEncryptor.encryptPassword(userpassword)
+        )
         TokenIssueResponse(jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES)))
     }
   }
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..71235669859
--- /dev/null
+++ b/amber/src/main/scala/org/apache/texera/web/resource/auth/ExternalAuthProvisioner.scala
@@ -0,0 +1,147 @@
+package org.apache.texera.web.resource.auth
+
+/*
+ * 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.
+ */
+
+import org.apache.texera.dao.SqlServer
+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 java.time.OffsetDateTime
+import scala.util.chaining.scalaUtilChainingOps
+
+/**
+  * A verified external identity (Google, Facebook, ...) reduced to the fields we
+  * persist. `avatar` is optional: `None` means the provider supplies no avatar, so
+  * the user's existing avatar column is left untouched rather than overwritten.
+  */
+final case class ExternalProfile(
+    providerType: ProviderTypeEnum,
+    providerId: String,
+    name: String,
+    email: String,
+    avatar: Option[String] = None
+)
+
+object ExternalAuthProvisioner {
+
+  /**
+    * Resolve the user behind an external identity, creating one if necessary, and
+    * ensure its auth-provider row is present and up to date. Runs in a single
+    * transaction and returns the (possibly newly created) user.
+    */
+  def loginOrProvision(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) =>
+          // known identity: refresh the profile fields if they drifted
+          txUserDao.fetchOneByUid(record.get(USER.UID)).tap { user =>
+            if (refresh(user, profile)) txUserDao.update(user)
+          }
+
+        case None =>
+          val user = Option(txUserDao.fetchOneByEmail(profile.email)) match {
+            case Some(existing) =>
+              existing.tap { user =>
+                if (refresh(user, profile)) txUserDao.update(user)
+              }
+            case None =>
+              new User().tap { user =>
+                user.setName(profile.name)
+                user.setEmail(profile.email)
+                profile.avatar.foreach(user.setAvatar)
+                user.setRole(UserRoleEnum.INACTIVE)
+                txUserDao.insert(user)
+              }
+          }
+
+          upsertProvider(ctx, txAuthDao, user, profile)
+          user
+      }
+    }
+
+  /**
+    * 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
+    }
+    profile.avatar.foreach { avatar =>
+      if (user.getAvatar != avatar) {
+        user.setAvatar(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 a088e5e56dd..1b6f085da95 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,30 +19,40 @@
 
 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.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
+
+  /**
+    * 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.
+    */
+  private[auth] def profileOf(payload: GoogleIdToken.Payload): ExternalProfile = {
+    val googleEmail = payload.getEmail
+    ExternalProfile(
+      ProviderTypeEnum.GOOGLE,
+      payload.getSubject,
+      Option(payload.get("name").asInstanceOf[String]).filter(_.nonEmpty).getOrElse(googleEmail),
+      googleEmail,
+      Some(
+        Option(payload.get("picture").asInstanceOf[String])
+          .flatMap(_.split("/").lastOption)
+          .getOrElse("")
+      )
     )
+  }
 }
 
 @Path("/auth/google")
@@ -53,64 +63,31 @@ class GoogleAuthResource {
   @Path("/clientid")
   def getClientId: String = clientId
 
-  @POST
-  @Consumes(Array(MediaType.TEXT_PLAIN))
-  @Produces(Array(MediaType.APPLICATION_JSON))
-  @Path("/login")
-  def login(credential: String): TokenIssueResponse = {
-    val idToken =
+  /**
+    * Verify `credential` against Google, yielding its payload, or None if it is not a valid
+    * token for this client. The only seam that reaches the network, so tests override it
+    * instead of signing a token; kept a method rather than a constructor parameter because
+    * Jersey instantiates this resource from `classOf[GoogleAuthResource]`.
+    */
+  protected def verifiedPayload(credential: String): Option[GoogleIdToken.Payload] =
+    Option(
       new GoogleIdTokenVerifier.Builder(new NetHttpTransport, GsonFactory.getDefaultInstance)
         .setAudience(
           Collections.singletonList(clientId)
         )
         .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(userDao.fetchOneByEmail(googleEmail)) match {
-            case Some(user) =>
-              if (user.getName != googleName) {
-                user.setName(googleName)
-              }
-              user.setGoogleId(googleId)
-              user.setGoogleAvatar(googleAvatar)
-              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.")
-  }
+    ).map(_.getPayload)
+
+  @POST
+  @Consumes(Array(MediaType.TEXT_PLAIN))
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @Path("/login")
+  def login(credential: String): TokenIssueResponse =
+    verifiedPayload(credential) match {
+      case Some(payload) =>
+        val user = ExternalAuthProvisioner.loginOrProvision(GoogleAuthResource.profileOf(payload))
+        TokenIssueResponse(jwtToken(jwtClaims(user, TOKEN_EXPIRE_TIME_IN_MINUTES)))
+      case None => throw new NotAuthorizedException("Login credentials are incorrect.")
+    }
 }
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 cd5ead915df..de4f9dd891f 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,18 +20,24 @@
 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.dao.jooq.generated.tables.daos.{AuthProviderDao, UserDao}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{AuthProvider, User}
 import org.apache.texera.web.resource.EmailTemplate.createRoleChangeTemplate
 import org.apache.texera.web.resource.GmailResource.sendEmail
-import org.apache.texera.web.resource.dashboard.admin.user.AdminUserResource.userDao
+import org.apache.texera.web.resource.dashboard.admin.user.AdminUserResource.{
+  passwordEncryptor,
+  userDao
+}
 import org.apache.texera.web.resource.dashboard.user.quota.UserQuotaResource._
 import org.jasypt.util.password.StrongPasswordEncryptor
+import org.jooq.exception.DataAccessException
 
 import java.util
+import java.util.UUID
 import javax.annotation.security.RolesAllowed
 import javax.ws.rs._
 import javax.ws.rs.core.{MediaType, Response}
@@ -41,8 +47,9 @@ case class UserInfo(
     name: String,
     email: String,
     googleId: String,
+    localHandle: String,
     role: UserRoleEnum,
-    googleAvatar: String,
+    avatar: String,
     comment: String,
     lastLogin: java.time.OffsetDateTime, // will be null if never logged in
     accountCreation: java.time.OffsetDateTime,
@@ -56,6 +63,8 @@ object AdminUserResource {
       .getInstance()
       .createDSLContext()
   private def userDao = new UserDao(context.configuration)
+  private val passwordEncryptor = new StrongPasswordEncryptor
+  private val UNIQUE_VIOLATION = "23505"
 }
 
 @Path("/admin/user")
@@ -71,14 +80,22 @@ class AdminUserResource {
   @Path("/list")
   @Produces(Array(MediaType.APPLICATION_JSON))
   def list(): util.List[UserInfo] = {
+
+    val googleProvider = AUTH_PROVIDER.as("google_provider")
+    val localProvider = AUTH_PROVIDER.as("local_provider")
+
     AdminUserResource.context
       .select(
         USER.UID,
         USER.NAME,
         USER.EMAIL,
-        USER.GOOGLE_ID,
+        // Both joins project a column called `provider_id`. fetchInto matches a case class by
+        // field NAME, so without these aliases the two collide and googleId silently receives
+        // whichever provider_id the mapper reaches first (the LOCAL handle).
+        googleProvider.PROVIDER_ID.as("googleId"),
+        localProvider.PROVIDER_ID.as("localHandle"),
         USER.ROLE,
-        USER.GOOGLE_AVATAR,
+        USER.AVATAR,
         USER.COMMENT,
         USER_LAST_ACTIVE_TIME.LAST_ACTIVE_TIME,
         USER.ACCOUNT_CREATION_TIME,
@@ -88,6 +105,12 @@ 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))
+      .leftJoin(localProvider)
+      .on(localProvider.PROVIDER_TYPE.eq(ProviderTypeEnum.LOCAL))
+      .and(localProvider.UID.eq(USER.UID))
       .fetchInto(classOf[UserInfo])
   }
 
@@ -116,12 +139,42 @@ 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)
+    val random = UUID.randomUUID().toString
+    createLocalAccount("User" + random, random)
+  }
+
+  /**
+    * Create a user together with the LOCAL credential it logs in with. Split out of `addUser`
+    * so the collision path is reachable: `addUser` derives its handle from a fresh UUID and
+    * so cannot produce the unique violation this maps to a 409.
+    */
+  private[user] def createLocalAccount(handle: String, rawPassword: String): Unit = {
+    val password = passwordEncryptor.encryptPassword(rawPassword)
+
+    try {
+      SqlServer.withTransaction(AdminUserResource.context) { ctx =>
+        val txUserDao = new UserDao(ctx.configuration())
+        val txAuthDao = new AuthProviderDao(ctx.configuration())
+
+        val newUser = new User
+        newUser.setName(handle)
+        newUser.setRole(UserRoleEnum.INACTIVE)
+        txUserDao.insert(newUser)
+
+        val newAuth = new AuthProvider()
+        newAuth.setUid(newUser.getUid)
+        newAuth.setPassword(password)
+        newAuth.setProviderType(ProviderTypeEnum.LOCAL)
+        newAuth.setProviderId(handle)
+        txAuthDao.insert(newAuth)
+      }
+    } catch {
+      case e: DataAccessException if e.sqlState() == AdminUserResource.UNIQUE_VIOLATION =>
+        throw new WebApplicationException(
+          new RuntimeException(s"Login handle $handle is already taken", e),
+          Response.Status.CONFLICT
+        )
+    }
   }
 
   @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/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/ExternalAuthProvisionerSpec.scala b/amber/src/test/scala/org/apache/texera/web/auth/ExternalAuthProvisionerSpec.scala
new file mode 100644
index 00000000000..8603069996e
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/auth/ExternalAuthProvisionerSpec.scala
@@ -0,0 +1,257 @@
+/*
+ * 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.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()
+
+  private def cleanup(): Unit =
+    getDSLContext.deleteFrom(USER).where(USER.EMAIL.like("%" + emailDomain)).execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  /** 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 = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(localPart + emailDomain)
+    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)
+
+  private def userCountByEmail(localPart: String): Int =
+    getDSLContext.fetchCount(USER, 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(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "google-sub-1",
+        "New User",
+        "new" + emailDomain,
+        Some("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 profile =
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "google-sub-return",
+        "Ret",
+        "ret" + emailDomain,
+        Some("a")
+      )
+
+    val first = ExternalAuthProvisioner.loginOrProvision(profile)
+    val second = ExternalAuthProvisioner.loginOrProvision(profile)
+
+    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(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-drift",
+        "Old Name",
+        "drift" + emailDomain,
+        Some("oldpic")
+      )
+    )
+    val updated = ExternalAuthProvisioner.loginOrProvision(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-drift",
+        "New Name",
+        "drift" + emailDomain,
+        Some("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(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-rename",
+        "Renamer",
+        "before" + emailDomain,
+        Some("pic")
+      )
+    )
+
+    val updated = ExternalAuthProvisioner.loginOrProvision(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-rename",
+        "Renamer",
+        "after" + emailDomain,
+        Some("pic")
+      )
+    )
+
+    updated.getUid shouldBe created.getUid
+    userDao.fetchOneByUid(created.getUid).getEmail shouldBe "after" + emailDomain
+    userCountByEmail("before") shouldBe 0
+  }
+
+  // `avatar = None` is the documented contract for providers that supply no picture: the
+  // column keeps whatever it held rather than being blanked on every login.
+  it should "leave the stored avatar untouched when the provider supplies none" in {
+    val existing = seedUser("Keeper", "keeper", avatar = "keep-me")
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-keeper",
+        "Keeper",
+        "keeper" + emailDomain,
+        None
+      )
+    )
+
+    result.getUid shouldBe existing.getUid
+    result.getAvatar shouldBe "keep-me"
+    userDao.fetchOneByUid(existing.getUid).getAvatar shouldBe "keep-me"
+  }
+
+  // ---- 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(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "sub-link",
+        "Local User",
+        "linkme" + emailDomain,
+        Some("pic")
+      )
+    )
+
+    result.getUid shouldBe existing.getUid
+    userCountByEmail("linkme") shouldBe 1
+    providerIdOf(existing.getUid, ProviderTypeEnum.GOOGLE) shouldBe "sub-link"
+  }
+
+  // ---- email match, provider row exists with a different id (upsert) --------
+
+  it should "update the existing provider id in place rather than inserting a colliding row" in {
+    val existing = seedUser("Rotating", "rotate")
+    seedExternalProvider(existing.getUid, ProviderTypeEnum.GOOGLE, "old-sub")
+
+    val result = ExternalAuthProvisioner.loginOrProvision(
+      ExternalProfile(
+        ProviderTypeEnum.GOOGLE,
+        "new-sub",
+        "Rotating",
+        "rotate" + emailDomain,
+        Some("p")
+      )
+    )
+
+    result.getUid shouldBe existing.getUid
+    providerRowCount(existing.getUid) shouldBe 1 // upserted, not a second row
+    providerIdOf(existing.getUid, ProviderTypeEnum.GOOGLE) shouldBe "new-sub"
+  }
+}
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..a435999aa1c 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,10 +35,9 @@ 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")
+    claims.setClaim("avatar", "avatar-blob")
     claims.setExpirationTimeMinutesInTheFuture(10f)
     claims
   }
@@ -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/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/auth/AuthResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala
new file mode 100644
index 00000000000..5f9d8b29553
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/AuthResourceSpec.scala
@@ -0,0 +1,379 @@
+/*
+ * 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.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.{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.jooq.exception.DataAccessException
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import javax.ws.rs.{NotAcceptableException, NotAuthorizedException}
+
+/**
+  * Integration spec for [[AuthResource]] against embedded Postgres ([[MockTexeraDB]] loads
+  * the real `texera_ddl.sql`, so `uq_provider_identity` / `ck_provider_credential` and the
+  * NOT NULL on `provider_id` are all exercised).
+  *
+  * The point of these tests is that the local login handle lives in
+  * `auth_provider.provider_id`, not in `"user".name` — so rewriting the display name (which
+  * an admin edit or a social login both do) must not disturb login.
+  */
+class AuthResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with MockTexeraDB {
+
+  // Every handle this spec registers starts with this, so cleanup can target the rows it
+  // made. Registration takes the email separately from the handle, so `register` derives one
+  // (see `emailFor`); cleanup keys off email because tests here deliberately rewrite `name`.
+  private val handlePrefix = "authspec_"
+
+  private var userDao: UserDao = _
+  private var authDao: AuthProviderDao = _
+  private var resource: AuthResource = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+    authDao = new AuthProviderDao(getDSLContext.configuration())
+    resource = new AuthResource()
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  override protected def beforeEach(): Unit = cleanup()
+  override protected def afterEach(): Unit = cleanup()
+
+  // The auth_provider FK is ON DELETE CASCADE, so deleting the user clears its rows.
+  // The bootstrap admin is not prefixed, so it needs deleting by its configured handle.
+  private def cleanup(): Unit =
+    getDSLContext
+      .deleteFrom(USER)
+      .where(
+        USER.EMAIL
+          .like(handlePrefix + "%")
+          .or(USER.EMAIL.eq(UserSystemConfig.adminUsername))
+      )
+      .execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  /** Registration validates the email's shape, so a bare handle cannot double as one. */
+  private def emailFor(handle: String): String = handle + "@example.com"
+
+  private def register(handle: String, password: String): Unit =
+    resource.register(UserRegistrationRequest(handle, emailFor(handle), password))
+
+  /** The LOCAL login handle recorded for a user, or null if they have no LOCAL row. */
+  private def localHandleOf(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)
+
+  private def userCountByEmail(email: String): Int =
+    getDSLContext.fetchCount(USER, USER.EMAIL.eq(email))
+
+  private def userByEmail(email: String): User = userDao.fetchOneByEmail(email)
+
+  /** The account `register(handle, _)` created, looked up the way cleanup keys off it. */
+  private def userByHandle(handle: String): User = userByEmail(emailFor(handle))
+
+  /** Rename the display name, the way an admin edit or a social-login refresh would. */
+  private def renameDisplayName(uid: Integer, newName: String): Unit = {
+    val user = userDao.fetchOneByUid(uid)
+    user.setName(newName)
+    userDao.update(user)
+  }
+
+  /** Seed a user backed by an external provider only — no LOCAL row, no password. */
+  private def seedExternalUser(name: String, email: String, providerId: String): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(email)
+    user.setRole(UserRoleEnum.REGULAR)
+    userDao.insert(user)
+
+    val auth = new AuthProvider
+    auth.setUid(user.getUid)
+    auth.setProviderType(ProviderTypeEnum.GOOGLE)
+    auth.setProviderId(providerId)
+    authDao.insert(auth)
+    user
+  }
+
+  // ---- registration and login round-trip -----------------------------------
+
+  behavior of "local registration and login"
+
+  it should "log in a freshly registered user and record the handle in auth_provider" in {
+    val handle = handlePrefix + "alice"
+    register(handle, "pw-alice")
+
+    val user = userByHandle(handle)
+    user should not be null
+    localHandleOf(user.getUid) shouldBe handle
+
+    val loggedIn = AuthResource.retrieveUserByUsernameAndPassword(handle, "pw-alice")
+    loggedIn.map(_.getUid) shouldBe Some(user.getUid)
+  }
+
+  it should "reject a wrong password" in {
+    val handle = handlePrefix + "bob"
+    register(handle, "pw-bob")
+
+    AuthResource.retrieveUserByUsernameAndPassword(handle, "not-pw-bob") shouldBe None
+  }
+
+  it should "reject an unknown handle" in {
+    AuthResource.retrieveUserByUsernameAndPassword(handlePrefix + "nobody", "pw") shouldBe None
+  }
+
+  // provider_id is only unique per provider_type, so an external subject id is not a login
+  // handle. Without the LOCAL predicate this matched a row whose password is null.
+  it should "refuse to treat an external provider's subject id as a login handle" in {
+    seedExternalUser("Külli", handlePrefix + "kulli@example.com", "google-sub-kulli")
+
+    AuthResource.retrieveUserByUsernameAndPassword("google-sub-kulli", "pw") shouldBe None
+  }
+
+  it should "log in the local account when an external identity shares its provider id" in {
+    val handle = handlePrefix + "liam"
+    register(handle, "pw-liam")
+    val uid = userByHandle(handle).getUid
+
+    // Same string as Liam's local handle, but registered under GOOGLE on another account —
+    // permitted by uq_provider_identity, and previously enough to make fetchOne() throw.
+    seedExternalUser("Liam Social", handlePrefix + "liam-social@example.com", handle)
+
+    AuthResource.retrieveUserByUsernameAndPassword(handle, "pw-liam").map(_.getUid) shouldBe Some(
+      uid
+    )
+  }
+
+  it should "reject null credentials without touching the database" in {
+    AuthResource.retrieveUserByUsernameAndPassword(null, "pw") shouldBe None
+    AuthResource.retrieveUserByUsernameAndPassword(handlePrefix + "alice", null) shouldBe None
+  }
+
+  it should "surface a bad login through the endpoint as 401" in {
+    val handle = handlePrefix + "carol"
+    register(handle, "pw-carol")
+
+    a[NotAuthorizedException] should be thrownBy
+      resource.login(UserLoginRequest(handle, "wrong"))
+  }
+
+  it should "refuse an empty or whitespace-only handle" in {
+    a[NotAcceptableException] should be thrownBy register("", "pw")
+    a[NotAcceptableException] should be thrownBy register("   ", "pw")
+  }
+
+  it should "refuse a null handle" in {
+    a[NotAcceptableException] should be thrownBy register(null, "pw")
+  }
+
+  it should "refuse a handle that is already taken" in {
+    val handle = handlePrefix + "dave"
+    register(handle, "pw-dave")
+
+    a[NotAcceptableException] should be thrownBy register(handle, "another-pw")
+  }
+
+  // ---- the regressions this design exists to prevent ------------------------
+
+  behavior of "the login handle's independence from the display name"
+
+  it should "keep login working after the display name is rewritten" in {
+    val handle = handlePrefix + "frank"
+    register(handle, "pw-frank")
+    val uid = userByHandle(handle).getUid
+
+    // exactly what AdminUserResource.updateUser and ExternalAuthProvisioner.refresh do
+    renameDisplayName(uid, "Frank The Renamed")
+
+    AuthResource.retrieveUserByUsernameAndPassword(handle, "pw-frank").map(_.getUid) shouldBe Some(
+      uid
+    )
+    localHandleOf(uid) shouldBe handle
+  }
+
+  it should "log in the right user when another account's display name collides with the handle" in {
+    val handle = handlePrefix + "grace"
+    register(handle, "pw-grace")
+    val uid = userByHandle(handle).getUid
+
+    // a social signup whose provider display name happens to equal Grace's login handle
+    seedExternalUser(handle, handlePrefix + "grace-social@example.com", "google-sub-grace")
+
+    AuthResource.retrieveUserByUsernameAndPassword(handle, "pw-grace").map(_.getUid) shouldBe Some(
+      uid
+    )
+  }
+
+  // ---- admin bootstrap -----------------------------------------------------
+
+  behavior of "createAdminUser"
+
+  it should "create exactly one admin and let it log in" in {
+    AuthResource.createAdminUser()
+
+    userCountByEmail(UserSystemConfig.adminUsername) shouldBe 1
+    val admin = userByEmail(UserSystemConfig.adminUsername)
+    admin.getRole shouldBe UserRoleEnum.ADMIN
+    localHandleOf(admin.getUid) shouldBe UserSystemConfig.adminUsername
+    AuthResource.retrieveUserByUsernameAndPassword(
+      UserSystemConfig.adminUsername,
+      UserSystemConfig.adminPassword
+    ) should not be None
+  }
+
+  it should "be idempotent across restarts" in {
+    AuthResource.createAdminUser()
+    AuthResource.createAdminUser()
+
+    userCountByEmail(UserSystemConfig.adminUsername) shouldBe 1
+  }
+
+  it should "not create a second admin after the first one is renamed" in {
+    AuthResource.createAdminUser()
+    val uid = userByEmail(UserSystemConfig.adminUsername).getUid
+    renameDisplayName(uid, "Renamed Admin")
+
+    AuthResource.createAdminUser()
+
+    userCountByEmail(UserSystemConfig.adminUsername) shouldBe 1
+    localHandleOf(uid) shouldBe UserSystemConfig.adminUsername
+  }
+
+  // The case that actually pins idempotency to the *handle*: once both the display name and the
+  // email have been edited, neither a fetchByName nor an email lookup recognises the admin, so a
+  // name-based check tries to insert and dies on uq_provider_identity — taking the boot with it,
+  // since nothing above createAdminUser catches.
+  it should "stay idempotent after the admin's name and email are both changed" in {
+    AuthResource.createAdminUser()
+    val uid = userByEmail(UserSystemConfig.adminUsername).getUid
+
+    val renamed = userDao.fetchOneByUid(uid)
+    renamed.setName("Renamed Admin")
+    renamed.setEmail(handlePrefix + "renamed-admin@example.com")
+    userDao.update(renamed)
+
+    noException should be thrownBy AuthResource.createAdminUser()
+
+    // still exactly one account holding the admin handle, and it is the original one
+    getDSLContext.fetchCount(
+      AUTH_PROVIDER,
+      AUTH_PROVIDER.PROVIDER_TYPE
+        .eq(ProviderTypeEnum.LOCAL)
+        .and(AUTH_PROVIDER.PROVIDER_ID.eq(UserSystemConfig.adminUsername))
+    ) shouldBe 1
+    localHandleOf(uid) shouldBe UserSystemConfig.adminUsername
+  }
+
+  // A deployment that leaves USER_SYS_ADMIN_USERNAME/PASSWORD blank gets no admin account at
+  // all, rather than one with an empty handle or an empty password.
+  it should "create nothing when the admin credentials are not configured" in {
+    AuthResource.createAdminUser("", "some-password")
+    AuthResource.createAdminUser(handlePrefix + "no-password-admin", "")
+
+    getDSLContext.fetchCount(USER, USER.EMAIL.like(handlePrefix + "%")) shouldBe 0
+    getDSLContext.fetchCount(AUTH_PROVIDER, AUTH_PROVIDER.PROVIDER_ID.eq("")) shouldBe 0
+  }
+
+  it should "not fail when the admin address is already held by an account with no local row" in {
+    // e.g. somebody signed in with Google using the configured admin address
+    seedExternalUser("Google Admin", UserSystemConfig.adminUsername, "google-sub-admin")
+
+    noException should be thrownBy AuthResource.createAdminUser()
+    userCountByEmail(UserSystemConfig.adminUsername) shouldBe 1
+  }
+
+  // ---- schema-level guarantees ---------------------------------------------
+
+  behavior of "the auth_provider constraints"
+
+  it should "refuse a LOCAL row with no handle" in {
+    val user = seedExternalUser(
+      "Handleless",
+      handlePrefix + "handleless@example.com",
+      "google-sub-handleless"
+    )
+
+    val auth = new AuthProvider
+    auth.setUid(user.getUid)
+    auth.setProviderType(ProviderTypeEnum.LOCAL)
+    auth.setPassword("hashed")
+    a[DataAccessException] should be thrownBy authDao.insert(auth)
+  }
+
+  it should "refuse two accounts sharing a LOCAL handle" in {
+    val handle = handlePrefix + "heidi"
+    register(handle, "pw-heidi")
+
+    val other =
+      seedExternalUser("Heidi Two", handlePrefix + "heidi2@example.com", "google-sub-heidi")
+    val auth = new AuthProvider
+    auth.setUid(other.getUid)
+    auth.setProviderType(ProviderTypeEnum.LOCAL)
+    auth.setProviderId(handle)
+    auth.setPassword("hashed")
+    a[DataAccessException] should be thrownBy authDao.insert(auth)
+  }
+
+  it should "refuse a LOCAL row with no password" in {
+    val user =
+      seedExternalUser("Ivan", handlePrefix + "ivan@example.com", "google-sub-ivan")
+
+    val passwordless = new AuthProvider
+    passwordless.setUid(user.getUid)
+    passwordless.setProviderType(ProviderTypeEnum.LOCAL)
+    passwordless.setProviderId(handlePrefix + "ivan-local")
+    a[DataAccessException] should be thrownBy authDao.insert(passwordless)
+  }
+
+  it should "refuse a non-LOCAL row that carries a password" in {
+    // A separate user: (uid, provider_type) is the primary key, so the GOOGLE slot on Ivan is
+    // already taken and the collision would be a PK violation rather than the check we want.
+    val user = new User
+    user.setName("Judy")
+    user.setEmail(handlePrefix + "judy@example.com")
+    user.setRole(UserRoleEnum.REGULAR)
+    userDao.insert(user)
+
+    val externalWithPassword = new AuthProvider
+    externalWithPassword.setUid(user.getUid)
+    externalWithPassword.setProviderType(ProviderTypeEnum.GOOGLE)
+    externalWithPassword.setProviderId("google-sub-judy")
+    externalWithPassword.setPassword("hashed")
+    a[DataAccessException] should be thrownBy authDao.insert(externalWithPassword)
+  }
+}
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..8572224343b
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/auth/GoogleAuthResourceSpec.scala
@@ -0,0 +1,185 @@
+/*
+ * 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 an unverifiable credential is a 401 rather than a crash.
+  */
+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.
+    */
+  private def payload(
+      subject: String,
+      email: String,
+      name: String = "Given Name",
+      picture: String = "https://lh3.googleusercontent.com/a/AVATAR-ID"
+  ): 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)
+    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")
+  }
+
+  // The one case that runs the real `verifiedPayload` rather than the stub. A string that is not
+  // three dot-separated parts fails in the local JWT parse, so no request to Google is made and
+  // this stays a unit test. Characterises today's behaviour: the parse error propagates, which
+  // Jersey renders as a 500 rather than the 401 a bad credential gets.
+  it should "fail on a malformed credential before reaching Google" in {
+    an[IllegalArgumentException] should be thrownBy new GoogleAuthResource().login("not-a-jwt")
+  }
+
+  // ---- 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/dashboard/DashboardResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DashboardResourceSpec.scala
new file mode 100644
index 00000000000..92494a33a40
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/DashboardResourceSpec.scala
@@ -0,0 +1,115 @@
+/*
+ * 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.dashboard
+
+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.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+import scala.jdk.CollectionConverters._
+
+/**
+  * Spec for [[DashboardResource.resultsOwnersInfo]] against embedded Postgres.
+  *
+  * The avatar it returns is read straight out of `"user".avatar` (formerly `google_avatar`), so
+  * the point here is that the column still round-trips: a rename that compiles but selects the
+  * wrong column would hand the frontend a null avatar for every owner instead of failing.
+  */
+class DashboardResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with MockTexeraDB {
+
+  private val emailDomain = "@dashboard-test.com"
+
+  private var userDao: UserDao = _
+  private var resource: DashboardResource = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+    resource = new DashboardResource()
+  }
+
+  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()
+
+  private def seedUser(name: String, localPart: String, avatar: String = null): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(localPart + emailDomain)
+    user.setRole(UserRoleEnum.REGULAR)
+    if (avatar != null) user.setAvatar(avatar)
+    userDao.insert(user)
+    user
+  }
+
+  private def ownersInfo(uids: Integer*): Map[Integer, DashboardResource.UserInfo] =
+    resource.resultsOwnersInfo(uids.asJava).asScala.toMap
+
+  behavior of "resultsOwnersInfo"
+
+  it should "return each owner's name and avatar" in {
+    val withAvatar = seedUser("Has Avatar", "has-avatar", avatar = "AVATAR-ID")
+
+    val info = ownersInfo(withAvatar.getUid).apply(withAvatar.getUid)
+
+    info.userId shouldBe withAvatar.getUid
+    info.userName shouldBe "Has Avatar"
+    info.googleAvatar shouldBe Some("AVATAR-ID")
+  }
+
+  it should "report no avatar for an owner that has none" in {
+    val without = seedUser("No Avatar", "no-avatar")
+
+    ownersInfo(without.getUid).apply(without.getUid).googleAvatar shouldBe None
+  }
+
+  it should "return one entry per requested owner" in {
+    val first = seedUser("First", "first", avatar = "one")
+    val second = seedUser("Second", "second", avatar = "two")
+
+    val info = ownersInfo(first.getUid, second.getUid)
+
+    info.keySet shouldBe Set(first.getUid, second.getUid)
+    info(first.getUid).googleAvatar shouldBe Some("one")
+    info(second.getUid).googleAvatar shouldBe Some("two")
+  }
+
+  it should "skip a uid that matches no user rather than failing" in {
+    val known = seedUser("Known", "known")
+
+    val info = ownersInfo(known.getUid, Integer.valueOf(-1))
+
+    info.keySet shouldBe Set(known.getUid)
+  }
+}
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
new file mode 100644
index 00000000000..48df94b0c88
--- /dev/null
+++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/admin/user/AdminUserResourceSpec.scala
@@ -0,0 +1,210 @@
+/*
+ * 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.dashboard.admin.user
+
+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.exception.DataAccessException
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, LoneElement}
+
+import javax.ws.rs.WebApplicationException
+import javax.ws.rs.core.Response
+import scala.jdk.CollectionConverters._
+
+/**
+  * Integration spec for [[AdminUserResource]] against embedded Postgres.
+  *
+  * Two things are worth pinning down here: the local login handle an admin-created account
+  * gets (it is no longer readable off `"user".name`, which an admin may rename), and the fact
+  * that `list()` joins `auth_provider` twice — once per provider — so the two handles must
+  * not bleed into each other's column.
+  */
+class AdminUserResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach
+    with LoneElement
+    with MockTexeraDB {
+
+  private var userDao: UserDao = _
+  private var authDao: AuthProviderDao = _
+  private var resource: AdminUserResource = _
+
+  override protected def beforeAll(): Unit = {
+    initializeDBAndReplaceDSLContext()
+    userDao = new UserDao(getDSLContext.configuration())
+    authDao = new AuthProviderDao(getDSLContext.configuration())
+    resource = new AdminUserResource()
+  }
+
+  override protected def afterAll(): Unit = shutdownDB()
+
+  override protected def beforeEach(): Unit = cleanup()
+  override protected def afterEach(): Unit = cleanup()
+
+  // This suite owns its embedded database, so clearing "user" is enough; auth_provider
+  // rows go with it via ON DELETE CASCADE.
+  private def cleanup(): Unit = getDSLContext.deleteFrom(USER).execute()
+
+  // ---- helpers -------------------------------------------------------------
+
+  private def seedUser(name: String, email: String): User = {
+    val user = new User
+    user.setName(name)
+    user.setEmail(email)
+    user.setRole(UserRoleEnum.REGULAR)
+    userDao.insert(user)
+    user
+  }
+
+  private def seedProvider(
+      uid: Integer,
+      providerType: ProviderTypeEnum,
+      providerId: String,
+      password: String = null
+  ): Unit = {
+    val auth = new AuthProvider
+    auth.setUid(uid)
+    auth.setProviderType(providerType)
+    auth.setProviderId(providerId)
+    if (password != null) auth.setPassword(password)
+    authDao.insert(auth)
+  }
+
+  private def infoFor(uid: Integer): UserInfo =
+    resource.list().asScala.find(_.uid == uid).getOrElse(fail(s"uid $uid missing from list()"))
+
+  private def localHandleOf(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)
+
+  private def allUsers: Seq[User] = userDao.findAll().asScala.toSeq
+
+  // ---- addUser -------------------------------------------------------------
+
+  behavior of "addUser"
+
+  it should "give the new account a local handle matching its generated name" in {
+    resource.addUser()
+
+    val created = allUsers.loneElement
+    created.getRole shouldBe UserRoleEnum.INACTIVE
+    localHandleOf(created.getUid) shouldBe created.getName
+  }
+
+  it should "generate distinct handles for accounts created in quick succession" in {
+    resource.addUser()
+    resource.addUser()
+    resource.addUser()
+
+    val handles = allUsers.map(u => localHandleOf(u.getUid))
+    handles should have size 3
+    handles.distinct should have size 3
+    handles.foreach(_ should not be null)
+  }
+
+  it should "leave the handle alone when the display name is later changed" in {
+    resource.addUser()
+    val created = allUsers.loneElement
+    val handle = created.getName
+
+    val renamed = userDao.fetchOneByUid(created.getUid)
+    renamed.setName("Friendly Name")
+    renamed.setEmail("friendly@example.com")
+    resource.updateUser(renamed)
+
+    localHandleOf(created.getUid) shouldBe handle
+    infoFor(created.getUid).name shouldBe "Friendly Name"
+    infoFor(created.getUid).localHandle shouldBe handle
+  }
+
+  // `addUser` cannot collide (its handle is a fresh UUID), so the mapping is driven through
+  // `createLocalAccount` directly — otherwise a taken handle would surface as a raw 500.
+  it should "reject a handle that is already taken with a 409 rather than a 500" in {
+    resource.createLocalAccount("taken-handle", "pw")
+
+    val thrown = the[WebApplicationException] thrownBy
+      resource.createLocalAccount("taken-handle", "another-pw")
+
+    thrown.getResponse.getStatus shouldBe Response.Status.CONFLICT.getStatusCode
+    allUsers should have size 1
+  }
+
+  // Only unique_violation means "handle taken"; every other database error must keep its own
+  // identity instead of being relabelled a conflict.
+  it should "let a database error that is not a unique violation propagate unchanged" in {
+    // provider_id is VARCHAR(256), so an over-long handle fails with 22001, not 23505
+    a[DataAccessException] should be thrownBy resource.createLocalAccount("h" * 300, "pw")
+  }
+
+  // ---- list ----------------------------------------------------------------
+
+  behavior of "list"
+
+  it should "report the local handle for a password account" in {
+    val user = seedUser("Local Only", "local@example.com")
+    seedProvider(user.getUid, ProviderTypeEnum.LOCAL, "local-handle", password = "hashed")
+
+    val info = infoFor(user.getUid)
+    info.localHandle shouldBe "local-handle"
+    info.googleId shouldBe null
+  }
+
+  it should "report the google id for an external account" in {
+    val user = seedUser("Google Only", "google@example.com")
+    seedProvider(user.getUid, ProviderTypeEnum.GOOGLE, "google-sub-1")
+
+    val info = infoFor(user.getUid)
+    info.googleId shouldBe "google-sub-1"
+    info.localHandle shouldBe null
+  }
+
+  it should "keep the two handles in their own columns for an account holding both" in {
+    val user = seedUser("Both", "both@example.com")
+    seedProvider(user.getUid, ProviderTypeEnum.LOCAL, "both-handle", password = "hashed")
+    seedProvider(user.getUid, ProviderTypeEnum.GOOGLE, "google-sub-2")
+
+    val info = infoFor(user.getUid)
+    info.localHandle shouldBe "both-handle"
+    info.googleId shouldBe "google-sub-2"
+    info.name shouldBe "Both"
+    info.email shouldBe "both@example.com"
+  }
+
+  // Two left joins against the same table are what make this a risk: without the
+  // provider_type predicates a user with several providers would fan out into one row each.
+  it should "return one row per user even when a user has several providers" in {
+    val user = seedUser("Both", "both@example.com")
+    seedProvider(user.getUid, ProviderTypeEnum.LOCAL, "both-handle-2", password = "hashed")
+    seedProvider(user.getUid, ProviderTypeEnum.GOOGLE, "google-sub-3")
+
+    resource.list().asScala.count(_.uid == user.getUid) shouldBe 1
+  }
+}
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 78aa85316ea..4338638323e 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
@@ -63,7 +63,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
@@ -75,7 +74,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
@@ -213,7 +211,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)
@@ -264,7 +261,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/user/project/ProjectAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/project/ProjectAccessResourceSpec.scala
index 2100be4b423..44f23831632 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
@@ -74,7 +74,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/workflow/WorkflowAccessResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowAccessResourceSpec.scala
index 72237b8befb..b795976751b 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
@@ -82,25 +82,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
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 97bbe34abc3..998ec453b0d 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
@@ -87,8 +87,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)
@@ -778,7 +777,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
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/pythonvirtualenvironment/PveResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala
index 9904e3eb354..4909d76648c 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/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..5b3364b69ac 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
@@ -55,10 +55,9 @@ object JwtAuth {
     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("avatar", user.getAvatar)
     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..bdda679ae87 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,7 @@ 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])
+    val googleAvatar = claims.getClaimValue("avatar", classOf[String])
 
     new SessionUser(
       new User().tap { user =>
@@ -71,8 +70,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..27a25abd0fa 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,8 +33,7 @@ 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
   }
@@ -43,12 +42,21 @@ class JwtAuthSpec extends AnyFlatSpec with Matchers {
     val claims = JwtAuth.jwtClaims(buildUser(), 7)
     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("avatar") shouldBe "avatar-blob"
     claims.getClaimValueAsString("role") shouldBe UserRoleEnum.ADMIN.name
   }
 
+  // Credentials now live in auth_provider, and the token deliberately carries none of them:
+  // it identifies the user, it does not re-present the identity that authenticated them.
+  it should "not carry any provider credential in the claims" in {
+    val claims = JwtAuth.jwtClaims(buildUser(), 7)
+    claims.hasClaim("googleId") shouldBe false
+    claims.hasClaim("googleAvatar") shouldBe false
+    claims.hasClaim("password") shouldBe false
+    claims.hasClaim("providerId") shouldBe false
+  }
+
   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 = {
@@ -68,8 +76,7 @@ class JwtAuthSpec extends AnyFlatSpec with Matchers {
     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
   }
 
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..ca979147c62 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,27 +38,24 @@ 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")
+    claims.setClaim("avatar", "avatar-blob")
     claims.setExpirationTimeMinutesInTheFuture(10f)
     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,10 +156,9 @@ 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")
+    bob.setClaim("avatar", "bob-avatar")
     bob.setExpirationTimeMinutesInTheFuture(10f)
 
     val aliceUser = JwtParser.parseToken(JwtAuth.jwtToken(alice)).get().getUser
@@ -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/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 aa02f73387e..a43298b0c05 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
@@ -430,7 +430,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
 
@@ -536,7 +536,7 @@ class ComputingUnitManagingResource {
           .fetchByUid(ownerUids: _*)
           .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)
           }
@@ -607,7 +607,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/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 4b5c78654ef..773c1e8b87d 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
new file mode 100644
index 00000000000..caaf681bbf0
--- /dev/null
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitManagingResourceSpec.scala
@@ -0,0 +1,190 @@
+/*
+ * 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.service.resource
+
+import jakarta.ws.rs.NotFoundException
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.Tables.COMPUTING_UNIT_USER_ACCESS
+import org.apache.texera.dao.jooq.generated.enums.{
+  PrivilegeEnum,
+  UserRoleEnum,
+  WorkflowComputingUnitTypeEnum
+}
+import org.apache.texera.dao.jooq.generated.tables.daos.{
+  ComputingUnitUserAccessDao,
+  UserDao,
+  WorkflowComputingUnitDao
+}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{
+  ComputingUnitUserAccess,
+  User,
+  WorkflowComputingUnit
+}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+
+/**
+  * Spec for [[ComputingUnitManagingResource.getComputingUnitInfo]], backed by embedded Postgres
+  * (via [[MockTexeraDB]]). Units are `local`, which keeps the status and metrics helpers away
+  * from Kubernetes — `local` is always Running with NaN metrics.
+  *
+  * The owner's avatar is read off `"user".avatar` (formerly `google_avatar`) and normalised so
+  * that "no avatar" reaches the frontend as null rather than an empty string; both halves of
+  * that are pinned here, since a wrong column would silently null every owner's avatar.
+  */
+class ComputingUnitManagingResourceSpec
+    extends AnyFlatSpec
+    with Matchers
+    with MockTexeraDB
+    with BeforeAndAfterAll
+    with BeforeAndAfterEach {
+
+  private val ownerUser: User = {
+    val user = new User
+    user.setName("info_owner")
+    user.setEmail("info_owner@test.com")
+    user.setRole(UserRoleEnum.REGULAR)
+    user.setAvatar("OWNER-AVATAR-ID")
+    user
+  }
+
+  private val blankAvatarUser: User = {
+    val user = new User
+    user.setName("info_blank_avatar")
+    user.setEmail("info_blank_avatar@test.com")
+    user.setRole(UserRoleEnum.REGULAR)
+    user.setAvatar("")
+    user
+  }
+
+  private val strangerUser: User = {
+    val user = new User
+    user.setName("info_stranger")
+    user.setEmail("info_stranger@test.com")
+    user.setRole(UserRoleEnum.REGULAR)
+    user
+  }
+
+  private val ownedUnit: WorkflowComputingUnit = {
+    val unit = new WorkflowComputingUnit
+    unit.setName("info-unit")
+    unit.setType(WorkflowComputingUnitTypeEnum.local)
+    unit.setUri("")
+    unit
+  }
+
+  private val blankAvatarUnit: WorkflowComputingUnit = {
+    val unit = new WorkflowComputingUnit
+    unit.setName("info-unit-blank-avatar")
+    unit.setType(WorkflowComputingUnitTypeEnum.local)
+    unit.setUri("")
+    unit
+  }
+
+  private lazy val resource = new ComputingUnitManagingResource()
+
+  private lazy val ownerSession = new SessionUser(ownerUser)
+  private lazy val strangerSession = new SessionUser(strangerUser)
+
+  override protected def beforeAll(): Unit = {
+    super.beforeAll()
+    initializeDBAndReplaceDSLContext()
+
+    val userDao = new UserDao(getDSLContext.configuration())
+    userDao.insert(ownerUser)
+    userDao.insert(blankAvatarUser)
+    userDao.insert(strangerUser)
+
+    val wcDao = new WorkflowComputingUnitDao(getDSLContext.configuration())
+    ownedUnit.setUid(ownerUser.getUid)
+    wcDao.insert(ownedUnit)
+    blankAvatarUnit.setUid(blankAvatarUser.getUid)
+    wcDao.insert(blankAvatarUnit)
+  }
+
+  override protected def beforeEach(): Unit = {
+    super.beforeEach()
+    // every test starts with no explicit grants
+    getDSLContext.deleteFrom(COMPUTING_UNIT_USER_ACCESS).execute()
+  }
+
+  override protected def afterAll(): Unit = {
+    try shutdownDB()
+    finally super.afterAll()
+  }
+
+  private def grantDirectly(cuid: Integer, uid: Integer, privilege: PrivilegeEnum): Unit = {
+    val access = new ComputingUnitUserAccess
+    access.setCuid(cuid)
+    access.setUid(uid)
+    access.setPrivilege(privilege)
+    new ComputingUnitUserAccessDao(getDSLContext.configuration()).insert(access)
+  }
+
+  behavior of "getComputingUnitInfo"
+
+  it should "report the owner's name and avatar" in {
+    val info = resource.getComputingUnitInfo(ownedUnit.getCuid, ownerSession)
+
+    info.ownerName shouldBe "info_owner"
+    info.ownerGoogleAvatar shouldBe "OWNER-AVATAR-ID"
+    info.computingUnit.getCuid shouldBe ownedUnit.getCuid
+  }
+
+  // Empty and absent are the same thing to the frontend, so an empty column becomes null
+  // rather than being passed through as "".
+  it should "report no avatar as null rather than an empty string" in {
+    val info = resource.getComputingUnitInfo(blankAvatarUnit.getCuid, ownerSession)
+
+    info.ownerGoogleAvatar shouldBe null
+    info.ownerName shouldBe "info_blank_avatar"
+  }
+
+  it should "grant the owner write access without an explicit access row" in {
+    val info = resource.getComputingUnitInfo(ownedUnit.getCuid, ownerSession)
+
+    info.isOwner shouldBe true
+    info.accessPrivilege shouldBe PrivilegeEnum.WRITE
+    info.status shouldBe "Running"
+  }
+
+  it should "report no privilege for a stranger with no access row" in {
+    val info = resource.getComputingUnitInfo(ownedUnit.getCuid, strangerSession)
+
+    info.isOwner shouldBe false
+    info.accessPrivilege shouldBe PrivilegeEnum.NONE
+  }
+
+  it should "report the granted privilege for a non-owner that has one" in {
+    grantDirectly(ownedUnit.getCuid, strangerUser.getUid, PrivilegeEnum.READ)
+
+    val info = resource.getComputingUnitInfo(ownedUnit.getCuid, strangerSession)
+
+    info.isOwner shouldBe false
+    info.accessPrivilege shouldBe PrivilegeEnum.READ
+  }
+
+  it should "reject a cuid that does not exist" in {
+    a[NotFoundException] should be thrownBy
+      resource.getComputingUnitInfo(Integer.valueOf(999999), ownerSession)
+  }
+}
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..b9d78fb48c9 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,7 +103,6 @@ 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))
   }
@@ -113,7 +112,6 @@ 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))
   }
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 78fc6f09c55..7587f06d6b7 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 fb27fa67319..bc975dbc4d1 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
@@ -137,7 +137,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 +145,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 +154,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
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 35c3f473773..95eee0166a8 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 2cec53da2f0..bdeb20bb5a9 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -58,6 +58,11 @@
         <sqlFile path="sql/updates/29.sql"/>
     </changeSet>
 
+    <!-- Split auth credentials out of "user" into auth_provider -->
+    <changeSet id="30" author="Neilk1021">
+        <sqlFile path="sql/updates/30.sql"/>
+    </changeSet>
+
     <!-- example changeSet
     <changeSet id="1" author="author">
         <sqlFile path="sql/updates/1.sql"/>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index 4132aba32c1..d1624b58018 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -78,6 +78,7 @@ DROP TABLE IF EXISTS computing_unit_user_access CASCADE;
 DROP TABLE IF EXISTS notebook CASCADE;
 DROP TABLE IF EXISTS workflow_notebook_mapping CASCADE;
 DROP TABLE IF EXISTS virtual_environments CASCADE;
+DROP TYPE IF EXISTS provider_type_enum CASCADE;
 
 -- ============================================
 -- 4. Create PostgreSQL enum types
@@ -91,6 +92,7 @@ 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');
 
 -- ============================================
 -- 5. Create tables
@@ -102,16 +104,25 @@ 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),
-    -- check that either password or google_id is not null
-    CONSTRAINT ck_nulltest CHECK ((password IS NOT NULL) OR (google_id IS NOT NULL))
+    joining_reason          VARCHAR(500)
+    );
+
+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),
+    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))
     );
 
 -- user_config
diff --git a/sql/updates/30.sql b/sql/updates/30.sql
new file mode 100644
index 00000000000..2f75535aa61
--- /dev/null
+++ b/sql/updates/30.sql
@@ -0,0 +1,154 @@
+/*
+ * 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.
+ */
+
+\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
+$$;
+
+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,
+
+    CONSTRAINT uq_provider_identity UNIQUE (provider_type, provider_id)
+    );
+
+ALTER TABLE auth_provider DROP CONSTRAINT IF EXISTS ck_provider_credential;
+
+DO $$
+DECLARE
+offenders TEXT;
+    orphans   TEXT;
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'password'
+    ) THEN
+        -- upgrading from the pre-auth_provider schema: handles come straight from "user"
+SELECT string_agg(DISTINCT quote_literal(name), ', ')
+INTO offenders
+FROM "user"
+WHERE password IS NOT NULL
+  AND (btrim(name) = '' OR name <> btrim(name) OR name IN (
+    SELECT name FROM "user" WHERE password IS NOT NULL
+    GROUP BY name HAVING count(*) > 1));
+
+SELECT string_agg(uid::TEXT, ', ')
+INTO orphans
+FROM "user"
+WHERE password IS NULL AND google_id IS NULL;
+ELSE
+SELECT string_agg(DISTINCT quote_literal(u.name), ', ')
+INTO offenders
+FROM "user" u
+         JOIN auth_provider a ON a.uid = u.uid AND a.provider_type = 'LOCAL'
+WHERE a.provider_id IS NULL
+  AND (btrim(u.name) = '' OR u.name <> btrim(u.name) OR u.name IN (
+    SELECT u2.name
+    FROM "user" u2
+             JOIN auth_provider a2 ON a2.uid = u2.uid AND a2.provider_type = 'LOCAL'
+    WHERE a2.provider_id IS NULL
+    GROUP BY u2.name HAVING count(*) > 1));
+END IF;
+
+    IF offenders IS NOT NULL THEN
+        RAISE EXCEPTION 'migration 29: cannot promote "user".name to a login handle - '
+                        'the following names are duplicated, blank, or whitespace-padded: %. '
+                        'Resolve them and re-run.', offenders;
+END IF;
+
+    IF orphans IS NOT NULL THEN
+        RAISE NOTICE 'migration 29: 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
+$$;
+
+-- 5. Backfill.
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema = 'texera_db' AND table_name = 'user' AND column_name = 'password'
+    ) THEN
+        INSERT INTO auth_provider (uid, provider_type, provider_id, password)
+SELECT uid, 'LOCAL'::provider_type_enum, name, 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
+$$;
+
+-- Fill handles left NULL by an earlier version of this migration; a no-op otherwise.
+UPDATE auth_provider a
+SET provider_id = u.name
+    FROM "user" u
+WHERE u.uid = a.uid
+  AND a.provider_type = 'LOCAL'
+  AND a.provider_id IS NULL;
+
+-- 6. Every row now has a handle, so make it mandatory and restore the credential check
+--    in its new shape: a password exists for LOCAL and only for LOCAL.
+ALTER TABLE auth_provider ALTER COLUMN provider_id SET NOT NULL;
+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
+$$;
+
+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;
\ No newline at end of file