@@ -215,8 +215,10 @@ SessionImpl createSession() {
 spanner.getOptions().getDatabaseRole(),
 spanner.getOptions().getSessionLabels(),
 options);
-return new SessionImpl(
-spanner, session.getName(), session.getCreateTime(), session.getMultiplexed(), options);
+SessionReference sessionReference =
+new SessionReference(
+session.getName(), session.getCreateTime(), session.getMultiplexed(), options);
+return new SessionImpl(spanner, sessionReference);
 } catch (RuntimeException e) {
 span.setStatus(e);
 throw e;
@@ -248,7 +250,9 @@ void createMultiplexedSession(SessionConsumer consumer) {
 true);
 SessionImpl sessionImpl =
 new SessionImpl(
-spanner, session.getName(), session.getCreateTime(), session.getMultiplexed(), null);
+spanner,
+new SessionReference(
+session.getName(), session.getCreateTime(), session.getMultiplexed(), null));
 consumer.onSessionReady(sessionImpl);
 } catch (Throwable t) {
 span.setStatus(t);
@@ -348,10 +352,11 @@ private List<SessionImpl> internalBatchCreateSessions(
 res.add(
 new SessionImpl(
 spanner,
-session.getName(),
-session.getCreateTime(),
-session.getMultiplexed(),
-options));
+new SessionReference(
+session.getName(),
+session.getCreateTime(),
+session.getMultiplexed(),
+options)));
 }
 return res;
 } catch (RuntimeException e) {
@@ -367,6 +372,6 @@ SessionImpl sessionWithId(String name) {
 synchronized (this) {
 options = optionMap(SessionOption.channelHint(sessionChannelCounter++));
 }
-return new SessionImpl(spanner, name, options);
+return new SessionImpl(spanner, new SessionReference(name, options));
 }
 }

@@ -17,7 +17,6 @@
 package com.google.cloud.spanner;
 
 import static com.google.cloud.spanner.SpannerExceptionFactory.newSpannerException;
-import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.SettableApiFuture;
@@ -28,7 +27,6 @@
 import com.google.cloud.spanner.AbstractReadContext.SingleUseReadOnlyTransaction;
 import com.google.cloud.spanner.Options.TransactionOption;
 import com.google.cloud.spanner.Options.UpdateOption;
-import com.google.cloud.spanner.SessionClient.SessionId;
 import com.google.cloud.spanner.TransactionRunnerImpl.TransactionContextImpl;
 import com.google.cloud.spanner.spi.v1.SpannerRpc;
 import com.google.common.base.Ticker;
@@ -97,49 +95,23 @@ interface SessionTransaction {
 }
 
 private final SpannerImpl spanner;
-private final String name;
-private final DatabaseId databaseId;
+private final SessionReference sessionReference;
 private SessionTransaction activeTransaction;
-private final Map<SpannerRpc.Option, ?> options;
-private volatile Instant lastUseTime;
-@Nullable private final Instant createTime;
-private final boolean isMultiplexed;
 private ISpan currentSpan;
 
-SessionImpl(SpannerImpl spanner, String name, Map<SpannerRpc.Option, ?> options) {
+SessionImpl(SpannerImpl spanner, SessionReference sessionReference) {
 this.spanner = spanner;
 this.tracer = spanner.getTracer();
-this.options = options;
-this.name = checkNotNull(name);
-this.databaseId = SessionId.of(name).getDatabaseId();
-this.lastUseTime = Instant.now();
-this.createTime = null;
-this.isMultiplexed = false;
-}
-
-SessionImpl(
-SpannerImpl spanner,
-String name,
-com.google.protobuf.Timestamp createTime,
-boolean isMultiplexed,
-Map<SpannerRpc.Option, ?> options) {
-this.spanner = spanner;
-this.tracer = spanner.getTracer();
-this.options = options;
-this.name = checkNotNull(name);
-this.databaseId = SessionId.of(name).getDatabaseId();
-this.lastUseTime = Instant.now();
-this.createTime = convert(createTime);
-this.isMultiplexed = isMultiplexed;
+this.sessionReference = sessionReference;
 }
 
 @Override
 public String getName() {
-return name;
+return sessionReference.getName();
 }
 
 Map<SpannerRpc.Option, ?> getOptions() {
-return options;
+return sessionReference.getOptions();
 }
 
 void setCurrentSpan(ISpan span) {
@@ -151,19 +123,27 @@ ISpan getCurrentSpan() {
 }
 
 Instant getLastUseTime() {
-return lastUseTime;
+return sessionReference.getLastUseTime();
 }
 
 Instant getCreateTime() {
-return createTime;
+return sessionReference.getCreateTime();
 }
 
 boolean getIsMultiplexed() {
-return isMultiplexed;
+return sessionReference.getIsMultiplexed();
+}
+
+SessionReference getSessionReference() {
+return sessionReference;
 }
 
 void markUsed(Instant instant) {
-lastUseTime = instant;
+sessionReference.markUsed(instant);
+}
+
+public DatabaseId getDatabaseId() {
+return sessionReference.getDatabaseId();
 }
 
 @Override
@@ -211,7 +191,7 @@ public CommitResponse writeAtLeastOnceWithOptions(
 Options options = Options.fromTransactionOptions(transactionOptions);
 final CommitRequest.Builder requestBuilder =
 CommitRequest.newBuilder()
-.setSession(name)
+.setSession(getName())
 .setReturnCommitStats(options.withCommitStats())
 .addAllMutations(mutationsProto);
 
@@ -239,7 +219,7 @@ public CommitResponse writeAtLeastOnceWithOptions(
 ISpan span = tracer.spanBuilder(SpannerImpl.COMMIT);
 try (IScope s = tracer.withSpan(span)) {
 return SpannerRetryHelper.runTxWithRetriesOnAborted(
-() -> new CommitResponse(spanner.getRpc().commit(request, this.options)));
+() -> new CommitResponse(spanner.getRpc().commit(request, getOptions())));
 } catch (RuntimeException e) {
 span.setStatus(e);
 throw e;
@@ -271,7 +251,9 @@ public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
 List<BatchWriteRequest.MutationGroup> mutationGroupsProto =
 MutationGroup.toListProto(mutationGroups);
 final BatchWriteRequest.Builder requestBuilder =
-BatchWriteRequest.newBuilder().setSession(name).addAllMutationGroups(mutationGroupsProto);
+BatchWriteRequest.newBuilder()
+.setSession(getName())
+.addAllMutationGroups(mutationGroupsProto);
 RequestOptions batchWriteRequestOptions = getRequestOptions(transactionOptions);
 if (batchWriteRequestOptions != null) {
 requestBuilder.setRequestOptions(batchWriteRequestOptions);
@@ -282,7 +264,7 @@ public ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
 }
 ISpan span = tracer.spanBuilder(SpannerImpl.BATCH_WRITE);
 try (IScope s = tracer.withSpan(span)) {
-return spanner.getRpc().batchWriteAtLeastOnce(requestBuilder.build(), this.options);
+return spanner.getRpc().batchWriteAtLeastOnce(requestBuilder.build(), getOptions());
 } catch (Throwable e) {
 span.setStatus(e);
 throw SpannerExceptionFactory.newSpannerException(e);
@@ -303,7 +285,7 @@ public ReadContext singleUse(TimestampBound bound) {
 .setSession(this)
 .setTimestampBound(bound)
 .setRpc(spanner.getRpc())
-.setDefaultQueryOptions(spanner.getDefaultQueryOptions(databaseId))
+.setDefaultQueryOptions(spanner.getDefaultQueryOptions(getDatabaseId()))
 .setDefaultPrefetchChunks(spanner.getDefaultPrefetchChunks())
 .setDefaultDecodeMode(spanner.getDefaultDecodeMode())
 .setDefaultDirectedReadOptions(spanner.getOptions().getDirectedReadOptions())
@@ -325,7 +307,7 @@ public ReadOnlyTransaction singleUseReadOnlyTransaction(TimestampBound bound) {
 .setSession(this)
 .setTimestampBound(bound)
 .setRpc(spanner.getRpc())
-.setDefaultQueryOptions(spanner.getDefaultQueryOptions(databaseId))
+.setDefaultQueryOptions(spanner.getDefaultQueryOptions(getDatabaseId()))
 .setDefaultPrefetchChunks(spanner.getDefaultPrefetchChunks())
 .setDefaultDecodeMode(spanner.getDefaultDecodeMode())
 .setDefaultDirectedReadOptions(spanner.getOptions().getDirectedReadOptions())
@@ -347,7 +329,7 @@ public ReadOnlyTransaction readOnlyTransaction(TimestampBound bound) {
 .setSession(this)
 .setTimestampBound(bound)
 .setRpc(spanner.getRpc())
-.setDefaultQueryOptions(spanner.getDefaultQueryOptions(databaseId))
+.setDefaultQueryOptions(spanner.getDefaultQueryOptions(getDatabaseId()))
 .setDefaultPrefetchChunks(spanner.getDefaultPrefetchChunks())
 .setDefaultDecodeMode(spanner.getDefaultDecodeMode())
 .setDefaultDirectedReadOptions(spanner.getOptions().getDirectedReadOptions())
@@ -379,14 +361,14 @@ public AsyncTransactionManagerImpl transactionManagerAsync(TransactionOption...
 
 @Override
 public ApiFuture<Empty> asyncClose() {
-return spanner.getRpc().asyncDeleteSession(name, options);
+return spanner.getRpc().asyncDeleteSession(getName(), getOptions());
 }
 
 @Override
 public void close() {
 ISpan span = tracer.spanBuilder(SpannerImpl.DELETE_SESSION);
 try (IScope s = tracer.withSpan(span)) {
-spanner.getRpc().deleteSession(name, options);
+spanner.getRpc().deleteSession(getName(), getOptions());
 } catch (RuntimeException e) {
 span.setStatus(e);
 throw e;
@@ -400,11 +382,11 @@ ApiFuture<ByteString> beginTransactionAsync(Options transactionOptions, boolean
 final ISpan span = tracer.spanBuilder(SpannerImpl.BEGIN_TRANSACTION);
 final BeginTransactionRequest request =
 BeginTransactionRequest.newBuilder()
-.setSession(name)
+.setSession(getName())
 .setOptions(createReadWriteTransactionOptions(transactionOptions))
 .build();
 final ApiFuture<Transaction> requestFuture =
-spanner.getRpc().beginTransactionAsync(request, options, routeToLeader);
+spanner.getRpc().beginTransactionAsync(request, getOptions(), routeToLeader);
 requestFuture.addListener(
 () -> {
 try (IScope s = tracer.withSpan(span)) {
@@ -446,7 +428,7 @@ TransactionContextImpl newTransaction(Options options) {
 .setOptions(options)
 .setTrackTransactionStarter(spanner.getOptions().isTrackTransactionStarter())
 .setRpc(spanner.getRpc())
-.setDefaultQueryOptions(spanner.getDefaultQueryOptions(databaseId))
+.setDefaultQueryOptions(spanner.getDefaultQueryOptions(getDatabaseId()))
 .setDefaultPrefetchChunks(spanner.getDefaultPrefetchChunks())
 .setDefaultDecodeMode(spanner.getDefaultDecodeMode())
 .setSpan(currentSpan)
@@ -458,9 +440,11 @@ TransactionContextImpl newTransaction(Options options) {
 
 <T extends SessionTransaction> T setActive(@Nullable T ctx) {
 throwIfTransactionsPending();
-
-if (activeTransaction != null) {
-activeTransaction.invalidate();
+// multiplexed sessions support running concurrent transactions
+if (!getIsMultiplexed()) {
+if (activeTransaction != null) {
+activeTransaction.invalidate();
+}
 }
 activeTransaction = ctx;
 if (activeTransaction != null) {
@@ -472,11 +456,4 @@ <T extends SessionTransaction> T setActive(@Nullable T ctx) {
 TraceWrapper getTracer() {
 return tracer;
 }
-
-private Instant convert(com.google.protobuf.Timestamp timestamp) {
-if (timestamp == null) {
-return null;
-}
-return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos());
-}
 }