@@ -37,9 +37,16 @@ private enum SessionMode {
 READ_WRITE
 }
 
+@VisibleForTesting final String clientId;
 @VisibleForTesting final SessionPool pool;
 
+@VisibleForTesting
 DatabaseClientImpl(SessionPool pool) {
+this("", pool);
+}
+
+DatabaseClientImpl(String clientId, SessionPool pool) {
+this.clientId = clientId;
 this.pool = pool;
 }
 

@@ -23,6 +23,8 @@
 class MetricRegistryConstants {
 
 // The label keys are used to uniquely identify timeseries.
+private static final LabelKey CLIENT_ID =
+LabelKey.create("client_id", "User defined database client id");
 private static final LabelKey DATABASE = LabelKey.create("database", "Target database");
 private static final LabelKey INSTANCE_ID =
 LabelKey.create("instance_id", "Name of the instance");
@@ -33,10 +35,10 @@ class MetricRegistryConstants {
 private static final LabelValue UNSET_LABEL = LabelValue.create(null);
 
 static final ImmutableList<LabelKey> SPANNER_LABEL_KEYS =
-ImmutableList.of(DATABASE, INSTANCE_ID, LIBRARY_VERSION);
+ImmutableList.of(CLIENT_ID, DATABASE, INSTANCE_ID, LIBRARY_VERSION);
 
 static final ImmutableList<LabelValue> SPANNER_DEFAULT_LABEL_VALUES =
-ImmutableList.of(UNSET_LABEL, UNSET_LABEL, UNSET_LABEL);
+ImmutableList.of(UNSET_LABEL, UNSET_LABEL, UNSET_LABEL, UNSET_LABEL);
 
 /** Unit to represent counts. */
 static final String COUNT = "1";

@@ -62,6 +62,24 @@ class SpannerImpl extends BaseService<SpannerOptions> implements Spanner {
 static final String QUERY = "CloudSpannerOperation.ExecuteQuery";
 static final String READ = "CloudSpannerOperation.ExecuteRead";
 
+private static final Object CLIENT_ID_LOCK = new Object();
+
+@GuardedBy("CLIENT_ID_LOCK")
+private static final Map<DatabaseId, Long> CLIENT_IDS = new HashMap<>();
+
+private static String nextDatabaseClientId(DatabaseId databaseId) {
+synchronized (CLIENT_ID_LOCK) {
+Long id = CLIENT_IDS.get(databaseId);
+if (id == null) {
+id = 1L;
+} else {
+id++;
+}
+CLIENT_IDS.put(databaseId, id);
+return String.format("client-%d", id);
+}
+}
+
 private final SpannerRpc gapicRpc;
 
 @GuardedBy("this")
@@ -153,24 +171,26 @@ public DatabaseClient getDatabaseClient(DatabaseId db) {
 if (dbClients.containsKey(db)) {
 return dbClients.get(db);
 } else {
+String clientId = nextDatabaseClientId(db);
 List<LabelValue> labelValues =
 ImmutableList.of(
+LabelValue.create(clientId),
 LabelValue.create(db.getDatabase()),
 LabelValue.create(db.getInstanceId().getName()),
 LabelValue.create(GaxProperties.getLibraryVersion(getOptions().getClass())));
 SessionPool pool =
 SessionPool.createPool(
 getOptions(), SpannerImpl.this.getSessionClient(db), labelValues);
-DatabaseClientImpl dbClient = createDatabaseClient(pool);
+DatabaseClientImpl dbClient = createDatabaseClient(clientId, pool);
 dbClients.put(db, dbClient);
 return dbClient;
 }
 }
 }
 
 @VisibleForTesting
-DatabaseClientImpl createDatabaseClient(SessionPool pool) {
-return new DatabaseClientImpl(pool);
+DatabaseClientImpl createDatabaseClient(String clientId, SessionPool pool) {
+return new DatabaseClientImpl(clientId, pool);
 }
 
 @Override

@@ -45,8 +45,8 @@ private static class SpannerWithClosedSessionsImpl extends SpannerImpl {
 }
 
 @Override
-DatabaseClientImpl createDatabaseClient(SessionPool pool) {
-return new DatabaseClientWithClosedSessionImpl(pool);
+DatabaseClientImpl createDatabaseClient(String clientId, SessionPool pool) {
+return new DatabaseClientWithClosedSessionImpl(clientId, pool);
 }
 }
 
@@ -58,8 +58,8 @@ public static class DatabaseClientWithClosedSessionImpl extends DatabaseClientIm
 private boolean invalidateNextSession = false;
 private boolean allowReplacing = true;
 
-DatabaseClientWithClosedSessionImpl(SessionPool pool) {
-super(pool);
+DatabaseClientWithClosedSessionImpl(String clientId, SessionPool pool) {
+super(clientId, pool);
 }
 
 /** Invalidate the next session that is checked out from the pool. */

@@ -1579,6 +1579,7 @@ public void testSessionMetrics() throws Exception {
 FakeMetricRegistry metricRegistry = new FakeMetricRegistry();
 List<LabelValue> labelValues =
 Arrays.asList(
+LabelValue.create("client1"),
 LabelValue.create("database1"),
 LabelValue.create("instance1"),
 LabelValue.create("1.0.0"));

@@ -32,6 +32,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -185,6 +186,49 @@ public void testSpannerClosed() throws InterruptedException {
 spanner4.close();
 }
 
+@Test
+public void testClientId() {
+// Create a unique database id to be sure it has not yet been used in the lifetime of this JVM.
+String dbName =
+String.format("projects/p1/instances/i1/databases/%s", UUID.randomUUID().toString());
+DatabaseId db = DatabaseId.of(dbName);
+
+Mockito.when(spannerOptions.getTransportOptions())
+.thenReturn(GrpcTransportOptions.newBuilder().build());
+Mockito.when(spannerOptions.getSessionPoolOptions())
+.thenReturn(SessionPoolOptions.newBuilder().setMinSessions(0).build());
+
+DatabaseClientImpl databaseClient = (DatabaseClientImpl) impl.getDatabaseClient(db);
+assertThat(databaseClient.clientId).isEqualTo("client-1");
+
+// Get same db client again.
+DatabaseClientImpl databaseClient1 = (DatabaseClientImpl) impl.getDatabaseClient(db);
+assertThat(databaseClient1.clientId).isEqualTo(databaseClient.clientId);
+
+// Get a db client for a different database.
+String dbName2 =
+String.format("projects/p1/instances/i1/databases/%s", UUID.randomUUID().toString());
+DatabaseId db2 = DatabaseId.of(dbName2);
+DatabaseClientImpl databaseClient2 = (DatabaseClientImpl) impl.getDatabaseClient(db2);
+assertThat(databaseClient2.clientId).isEqualTo("client-1");
+
+// Create a new Spanner instance. This will generate new database clients with new ids.
+try (Spanner spanner =
+SpannerOptions.newBuilder()
+.setProjectId("p1")
+.setCredentials(NoCredentials.getInstance())
+.build()
+.getService()) {
+
+// Get a database client for the same database as the first database. As this goes through a
+// different Spanner instance with potentially different options, it will get a different
+// client
+// id.
+DatabaseClientImpl databaseClient3 = (DatabaseClientImpl) spanner.getDatabaseClient(db);
+assertThat(databaseClient3.clientId).isEqualTo("client-2");
+}
+}
+
 private SpannerOptions createSpannerOptions() {
 return SpannerOptions.newBuilder()
 .setProjectId("[PROJECT]")