diff --git a/docs/PER_SHARE_EGRESS_MONITORING.md b/docs/PER_SHARE_EGRESS_MONITORING.md
index e5dfa897d..4477be7aa 100644
--- a/docs/PER_SHARE_EGRESS_MONITORING.md
+++ b/docs/PER_SHARE_EGRESS_MONITORING.md
@@ -9,6 +9,7 @@ When enabled, each data access emits JSON logs with:
 - Total egress bytes
 - Pricing tier (e.g., `internet_to_na_eu`, `interregion_na_to_eu`)
 - Client region code
+- Audit fields for customer audits (clientIp, rawRegionHeader, isGcpIp, tenantId)
 
 ---
 
@@ -125,6 +126,12 @@ accessLogging:
   detectGcpTraffic: true                # Enable GCP IP range lookup
   clientRegionHeader: "x-client-region" # Header with country code
   clientIpHeader: "x-forwarded-for"     # Header with client IP chain
+  # Base path for the consolidated access log Delta table.
+  # All access logs are written to a single table: access_log_br__system
+  # Omit to disable Delta writing.
+  deltaTablePath: "gs://<bucket>/datalake/data/tenant/_system"
+  deltaFlushIntervalSeconds: 60         # Max seconds between Delta flushes (default: 60)
+  deltaFlushBatchSize: 1000             # Records per flush before early trigger (default: 1000)
 ```
 
 ### GCP Load Balancer Headers
@@ -137,6 +144,73 @@ X-Client-Region-Subdivision: {client_region_subdivision}
 
 ---
 
+## Delta Lake Storage
+
+When `deltaTablePath` is configured, `ACCESS_LOG` entries are written asynchronously to
+a consolidated Delta table on GCS in addition to the JSON log stream. This enables durable storage and
+SQL-queryable access via Delta Sharing.
+
+### Consolidated Table
+
+All access logs are written to a single consolidated table `access_log_br__system`. The `tenantId`
+field is included in each record for filtering by tenant. This simplifies cross-tenant queries
+and consolidates all egress data in one location.
+
+**Note:** `access_log_br__system` has double underscore in it's name (one as a separator and one from _system tenant's name).
+
+**IMPORTANT:** The Delta table must be pre-created by the `deltalake-admin` tool during tenant
+onboarding. The Delta Sharing server does not auto-create the table schema. 
+
+| Property | Value |
+|----------|-------|
+| **Table Name** | `access_log_br__system` |
+| **GCS Path** | `{deltaTablePath}/access_log_br__system` |
+| **Partitioning** | None (unpartitioned for simplified queries) |
+| **Format** | Parquet + Delta transaction log |
+| **Pre-requisite** | Table must be created by `deltalake-admin` before first write |
+
+### Schema
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `logType` | string | Always "ACCESS_LOG" |
+| `share` | string | Share name |
+| `schema` | string | Schema name |
+| `table` | string | Table name |
+| `egressBytes` | long | Bytes transferred |
+| `pricingTier` | string | GCP pricing tier (see Pricing Tiers) |
+| `timestampMs` | long | Timestamp in milliseconds |
+| `requestType` | string | "query" or "cdf_stream" |
+| `clientRegion` | string | ISO 3166-1 alpha-2 country code |
+| `tenantId` | string | Tenant identifier (derived from share name) |
+| `clientIp` | string | Client IP address for audit |
+| `rawRegionHeader` | string | Raw region header value for audit |
+| `isGcpIp` | boolean | Whether client IP is in GCP ranges |
+
+### GCS Base Paths by Environment
+
+| Environment | Table Path |
+|-------------|------------|
+| zing-dev | `gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zing-preview | `gs://zing-preview-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod | `gs://zcloud-prod-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod2 | `gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+| zcloud-prod3 | `gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system/access_log_br__system` |
+
+### Delta Write Behavior
+
+- Records are buffered in a bounded in-memory queue (capacity: 100,000) and written by a
+  background daemon thread — the request path is never blocked.
+- A flush is triggered when `deltaFlushBatchSize` records accumulate or `deltaFlushIntervalSeconds`
+  elapses, whichever comes first.
+- Queue overflow silently drops records (logged as a warning); write errors are logged to stderr
+  and never propagate to clients.
+- On graceful shutdown the writer makes a best-effort attempt to drain the queue (up to ~30s) before the process exits.
+- Only `ACCESS_LOG` entries are written to Delta. `PRICING_CONTEXT` and `REQUEST_HEADERS`
+  entries are log-only.
+
+---
+
 ## Log Output
 
 **ACCESS_LOG** — Emitted for each request with non-zero egress:
@@ -150,7 +224,11 @@ X-Client-Region-Subdivision: {client_region_subdivision}
   "pricingTier": "internet_to_na_eu",
   "timestampMs": 1717502400000,
   "requestType": "query",
-  "clientRegion": "US"
+  "clientRegion": "US",
+  "tenantId": "my_tenant",
+  "clientIp": "203.0.113.45",
+  "rawRegionHeader": "US",
+  "isGcpIp": false
 }
 ```
 
@@ -191,8 +269,9 @@ X-Client-Region-Subdivision: {client_region_subdivision}
 |------|---------|
 | `GcpPricingTier.scala` | Continent mapping, egress type detection, pricing calculation |
 | `GcpIpRangeLookup.scala` | GCP IP range fetching and CIDR trie lookup |
-| `AccessLogEmitter.scala` | Log entry models and JSON emission |
-| `DeltaSharingService.scala` | Integration for query/CDF endpoints |
+| `AccessLogEmitter.scala` | Log entry models with audit fields, JSON emission, composite fan-out |
+| `DeltaAccessLogWriter.scala` | Async Delta Lake writer to consolidated `access_log_br__system` table |
+| `DeltaSharingService.scala` | Integration for query/CDF endpoints, tenant ID extraction |
 | `ServerConfig.scala` | `AccessLoggingConfig` model |
 
 ### Egress Bytes Calculation
@@ -203,7 +282,10 @@ Sum of `size` from all file actions: `AddFile`, `AddFileForCDF`, `AddCDCFile`.
 
 ## Notes
 
-- Logs emitted to `delta.sharing.access` logger
-- Zero-byte requests not logged
+- JSON logs emitted to `delta.sharing.access` logger
+- Zero-byte requests not logged (neither to JSON nor Delta)
 - GCP IP ranges refreshed every 24 hours
 - Pricing tiers match GCP documentation
+- Delta writes are additive; disabling `deltaTablePath` has no effect on JSON log output
+- All tenant access logs are consolidated in a single `access_log_br__system` table
+- Audit fields (clientIp, rawRegionHeader, isGcpIp, tenantId) support customer audit requirements
diff --git a/manifests/base/configmap.yaml b/manifests/base/configmap.yaml
index e0112b525..889b7cb51 100644
--- a/manifests/base/configmap.yaml
+++ b/manifests/base/configmap.yaml
@@ -32,3 +32,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zing-dev-197522-dl-v1/datalake/data/tenant/_system"
diff --git a/manifests/zcloud-prod/configmap.yaml b/manifests/zcloud-prod/configmap.yaml
index 155fd5738..650c116fa 100644
--- a/manifests/zcloud-prod/configmap.yaml
+++ b/manifests/zcloud-prod/configmap.yaml
@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod-dl-v1/datalake/data/tenant/_system"
diff --git a/manifests/zcloud-prod2/configmap.yaml b/manifests/zcloud-prod2/configmap.yaml
index 577806dd9..ddb23c58d 100644
--- a/manifests/zcloud-prod2/configmap.yaml
+++ b/manifests/zcloud-prod2/configmap.yaml
@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod2-dl-v1/datalake/data/tenant/_system"
diff --git a/manifests/zcloud-prod3/configmap.yaml b/manifests/zcloud-prod3/configmap.yaml
index 62b112646..fa98e64a4 100644
--- a/manifests/zcloud-prod3/configmap.yaml
+++ b/manifests/zcloud-prod3/configmap.yaml
@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zcloud-prod3-dl-v1/datalake/data/tenant/_system"
diff --git a/manifests/zing-preview/configmap.yaml b/manifests/zing-preview/configmap.yaml
index 155fd5738..fdcf6e061 100644
--- a/manifests/zing-preview/configmap.yaml
+++ b/manifests/zing-preview/configmap.yaml
@@ -27,3 +27,4 @@ data:
       detectGcpTraffic: true
       clientRegionHeader: "x-client-region"
       clientIpHeader: "x-forwarded-for"
+      deltaTablePath: "gs://zing-preview-dl-v1/datalake/data/tenant/_system"
diff --git a/scalastyle-config.xml b/scalastyle-config.xml
index 59c1afb76..cf204534e 100644
--- a/scalastyle-config.xml
+++ b/scalastyle-config.xml
@@ -92,7 +92,7 @@ This file is divided into 3 sections:
  */
 
 \E)?\Q/*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (\E20\d{2}\Q) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala b/server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala
index 01bb426f3..65640d78f 100644
--- a/server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala
+++ b/server/src/main/scala/io/delta/sharing/server/DeltaSharingService.scala
@@ -207,6 +207,18 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       .toMap
   }
 
+  /**
+   * Extracts tenant_id from share name following the pattern `{tenant_id}_share`.
+   * Falls back to the share name itself if the pattern doesn't match.
+   */
+  private def extractTenantId(shareName: String): String = {
+    if (shareName.endsWith("_share")) {
+      shareName.dropRight("_share".length)
+    } else {
+      shareName
+    }
+  }
+
   private def emitQueryEgressMetric(
       req: HttpRequest,
       share: String,
@@ -228,7 +240,7 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       headers,
       serverConfig.getAccessLogging)
 
-    // Emit access log with essential fields
+    // Emit access log with essential fields and audit fields
     val entry = AccessLogEntry(
       share = share,
       schema = schema,
@@ -237,7 +249,11 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       timestampMs = nowMs,
       pricingTier = pricingCtx.location.pricingTier,
       clientRegion = pricingCtx.location.clientRegion,
-      requestType = AccessLogEmitter.QueryRequestType
+      requestType = AccessLogEmitter.QueryRequestType,
+      tenantId = Some(extractTenantId(share)),
+      clientIp = pricingCtx.clientIp,
+      rawRegionHeader = pricingCtx.rawRegionHeader,
+      isGcpIp = Some(pricingCtx.isGcpIp)
     )
     accessLogEmitter.record(entry)
 
@@ -292,7 +308,7 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       headers,
       serverConfig.getAccessLogging)
 
-    // Emit access log with essential fields
+    // Emit access log with essential fields and audit fields
     val entry = AccessLogEntry(
       share = share,
       schema = schema,
@@ -301,7 +317,11 @@ class DeltaSharingService(serverConfig: ServerConfig) {
       timestampMs = nowMs,
       pricingTier = pricingCtx.location.pricingTier,
       clientRegion = pricingCtx.location.clientRegion,
-      requestType = AccessLogEmitter.CdfStreamRequestType
+      requestType = AccessLogEmitter.CdfStreamRequestType,
+      tenantId = Some(extractTenantId(share)),
+      clientIp = pricingCtx.clientIp,
+      rawRegionHeader = pricingCtx.rawRegionHeader,
+      isGcpIp = Some(pricingCtx.isGcpIp)
     )
     accessLogEmitter.record(entry)
 
@@ -1220,6 +1240,12 @@ object DeltaSharingService {
   }
 
   def start(serverConfig: ServerConfig): Server = {
+    val service = new DeltaSharingService(serverConfig)
+    // scalastyle:off runtimeaddshutdownhook
+    Runtime.getRuntime.addShutdownHook(new Thread(
+      () => service.accessLogEmitter.close(),
+      "delta-access-log-shutdown"))
+    // scalastyle:on runtimeaddshutdownhook
     lazy val server = {
       updateDefaultJsonPrinterForScalaPbConverterUtil()
       val builder = Server.builder()
@@ -1227,7 +1253,7 @@ object DeltaSharingService {
         .disableDateHeader()
         .disableServerHeader()
         .requestTimeout(java.time.Duration.ofSeconds(serverConfig.requestTimeoutSeconds))
-        .annotatedService(serverConfig.endpoint, new DeltaSharingService(serverConfig): Any)
+        .annotatedService(serverConfig.endpoint, service: Any)
       if (serverConfig.ssl == null) {
         builder.http(serverConfig.getPort)
       } else {
diff --git a/server/src/main/scala/io/delta/sharing/server/config/ServerConfig.scala b/server/src/main/scala/io/delta/sharing/server/config/ServerConfig.scala
index 3cfd15df0..f1a6b2811 100644
--- a/server/src/main/scala/io/delta/sharing/server/config/ServerConfig.scala
+++ b/server/src/main/scala/io/delta/sharing/server/config/ServerConfig.scala
@@ -152,7 +152,17 @@ case class AccessLoggingConfig(
     // Enable GCP traffic detection using GCP's published IP ranges (cloud.json).
     // When true, client IPs belonging to other GCP regions can be classified as inter-region
     // (cheaper) rather than internet egress. Set to false to disable this detection.
-    @BeanProperty var detectGcpTraffic: Boolean) extends ConfigItem {
+    @BeanProperty var detectGcpTraffic: Boolean,
+    // GCS base path for the consolidated access log Delta table. When set, ACCESS_LOG
+    // entries are written asynchronously to `{deltaTablePath}/access_log_br__system`
+    // in addition to JSON logs. The Delta table must be pre-created by deltalake-admin;
+    // the server does not auto-create schema. Leave null or empty to disable Delta writing.
+    // Example: gs://my-bucket/datalake/data/tenant/_system
+    @BeanProperty var deltaTablePath: String,
+    // How often (seconds) to flush buffered access log records to the Delta table.
+    @BeanProperty var deltaFlushIntervalSeconds: Int,
+    // Maximum number of records to buffer before triggering an early flush.
+    @BeanProperty var deltaFlushBatchSize: Int) extends ConfigItem {
 
   def this() = {
     this(
@@ -161,7 +171,10 @@ case class AccessLoggingConfig(
       clientIpHeader = "x-forwarded-for",
       pricingGroups = Collections.emptyMap(),
       sourceRegion = "",
-      detectGcpTraffic = true)
+      detectGcpTraffic = true,
+      deltaTablePath = null,
+      deltaFlushIntervalSeconds = 60,
+      deltaFlushBatchSize = 1000)
   }
 
   override def checkConfig(): Unit = {
diff --git a/server/src/main/scala/io/delta/sharing/server/telemetry/AccessLogEmitter.scala b/server/src/main/scala/io/delta/sharing/server/telemetry/AccessLogEmitter.scala
index c9ee3ecb5..a209af2c1 100644
--- a/server/src/main/scala/io/delta/sharing/server/telemetry/AccessLogEmitter.scala
+++ b/server/src/main/scala/io/delta/sharing/server/telemetry/AccessLogEmitter.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -62,6 +62,12 @@ import io.delta.sharing.server.config.ServerConfig
  * == Optional Context ==
  * @param clientRegion ISO 3166-1 alpha-2 country code of the client (e.g., "US", "MT")
  * @param requestType Type of request: "query" (snapshot read) or "cdf_stream" (CDF streaming)
+ *
+ * == Audit Fields (for customer audits and consolidated storage) ==
+ * @param tenantId Tenant identifier extracted from share name (for consolidated table storage)
+ * @param clientIp Client IP address from request headers (for audit)
+ * @param rawRegionHeader Raw region header value before normalization (for audit)
+ * @param isGcpIp Whether the client IP is in a known GCP public IP range (for audit)
  */
 case class AccessLogEntry(
     share: String,
@@ -71,7 +77,11 @@ case class AccessLogEntry(
     timestampMs: Long,
     pricingTier: String = "unknown",
     clientRegion: Option[String] = None,
-    requestType: String = "query")
+    requestType: String = "query",
+    tenantId: Option[String] = None,
+    clientIp: Option[String] = None,
+    rawRegionHeader: Option[String] = None,
+    isGcpIp: Option[Boolean] = None)
 
 /**
  * Captures all context information used to calculate the pricing tier.
@@ -139,6 +149,7 @@ trait AccessLogEmitter {
   def record(entry: AccessLogEntry): Unit
   def recordContext(entry: PricingContextLogEntry): Unit
   def recordHeaders(entry: RequestHeadersLogEntry): Unit
+  def close(): Unit = {}
 }
 
 object AccessLogEmitter {
@@ -148,11 +159,24 @@ object AccessLogEmitter {
   /**
    * Creates an AccessLogEmitter based on server configuration.
    * Returns a JsonAccessLogEmitter if access logging is enabled, otherwise a no-op emitter.
+   * When deltaTablePath is also set, returns a CompositeAccessLogEmitter that fans out to
+   * both the JSON log stream and the Delta table writer.
    */
   def create(serverConfig: ServerConfig): AccessLogEmitter = {
     val cfg = Option(serverConfig.getAccessLogging)
     cfg match {
-      case Some(c) if c.enabled => new JsonAccessLogEmitter()
+      case Some(c) if c.enabled =>
+        val jsonEmitter = new JsonAccessLogEmitter()
+        val deltaPath = Option(c.getDeltaTablePath).map(_.trim).filter(_.nonEmpty)
+        deltaPath match {
+          case Some(path) =>
+            val deltaWriter = new DeltaAccessLogWriter(
+              path,
+              c.getDeltaFlushIntervalSeconds,
+              c.getDeltaFlushBatchSize)
+            new CompositeAccessLogEmitter(Seq(jsonEmitter, deltaWriter))
+          case None => jsonEmitter
+        }
       case _ => NoopAccessLogEmitter
     }
   }
@@ -167,6 +191,21 @@ object NoopAccessLogEmitter extends AccessLogEmitter {
   override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {}
 }
 
+/**
+ * Fans out all emitter calls to a sequence of delegate emitters.
+ * Used to write to both JSON logs and the Delta table simultaneously.
+ */
+class CompositeAccessLogEmitter(emitters: Seq[AccessLogEmitter]) extends AccessLogEmitter {
+  override def record(entry: AccessLogEntry): Unit =
+    emitters.foreach(_.record(entry))
+  override def recordContext(entry: PricingContextLogEntry): Unit =
+    emitters.foreach(_.recordContext(entry))
+  override def recordHeaders(entry: RequestHeadersLogEntry): Unit =
+    emitters.foreach(_.recordHeaders(entry))
+  override def close(): Unit =
+    emitters.foreach(_.close())
+}
+
 /**
  * Emits access log entries as JSON-structured log lines.
  * Uses a dedicated logger that can be filtered/routed separately in Cloud Logging.
@@ -192,9 +231,13 @@ class JsonAccessLogEmitter extends AccessLogEmitter {
       "requestType" -> entry.requestType
     )
 
-    // Optional context fields
+    // Optional context fields (including audit fields)
     val contextPayload = Seq(
-      entry.clientRegion.map("clientRegion" -> _)
+      entry.clientRegion.map("clientRegion" -> _),
+      entry.tenantId.map("tenantId" -> _),
+      entry.clientIp.map("clientIp" -> _),
+      entry.rawRegionHeader.map("rawRegionHeader" -> _),
+      entry.isGcpIp.map("isGcpIp" -> _)
     ).flatten.toMap
 
     val logPayload = basePayload ++ contextPayload
diff --git a/server/src/main/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriter.scala b/server/src/main/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriter.scala
new file mode 100644
index 000000000..cd4bf5ec6
--- /dev/null
+++ b/server/src/main/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriter.scala
@@ -0,0 +1,295 @@
+/*
+ * Copyright (2026) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.sharing.server.telemetry
+
+import java.util.UUID
+import java.util.concurrent.{LinkedBlockingQueue, TimeUnit}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+
+import io.delta.standalone.DeltaLog
+import io.delta.standalone.Operation
+import io.delta.standalone.actions.AddFile
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.example.data.simple.SimpleGroupFactory
+import org.apache.parquet.hadoop.ParquetFileWriter
+import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import org.apache.parquet.io.api.Binary
+import org.apache.parquet.schema.{MessageType, Types => PTypes}
+import org.apache.parquet.schema.LogicalTypeAnnotation
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.{BINARY, BOOLEAN, INT64}
+import org.slf4j.LoggerFactory
+
+/**
+ * Writes ACCESS_LOG entries asynchronously to a single consolidated Delta table on GCS.
+ *
+ * Records are buffered in a bounded in-memory queue and written by a single background
+ * daemon thread. The calling thread is never blocked: records are silently dropped when
+ * the queue is full. Write failures are logged to stderr and never propagate to callers.
+ *
+ * All access logs are written to a single table at `{basePath}/access_log_br__system`,
+ * with tenant_id included as a field for filtering. The table is not partitioned to
+ * simplify queries across all tenants.
+ *
+ * IMPORTANT: The Delta table must be pre-created by the deltalake-admin tool during
+ * tenant onboarding. This writer does not create the table schema.
+ *
+ * Only ACCESS_LOG entries are written; PRICING_CONTEXT and REQUEST_HEADERS are ignored.
+ *
+ * @param basePath GCS base path for the consolidated access log table
+ *        (e.g. gs://bucket/datalake/data/tenant/_system)
+ * @param flushIntervalSeconds how often to flush buffered records (seconds)
+ * @param flushBatchSize maximum records per flush (triggers early flush when reached)
+ */
+class DeltaAccessLogWriter(
+    basePath: String,
+    flushIntervalSeconds: Int,
+    flushBatchSize: Int) extends AccessLogEmitter {
+
+  private val logger = LoggerFactory.getLogger(classOf[DeltaAccessLogWriter])
+
+  private val MaxQueueCapacity = 100000
+  private val queue = new LinkedBlockingQueue[AccessLogEntry](MaxQueueCapacity)
+  private val stopped = new AtomicBoolean(false)
+  private val flushIntervalMs = flushIntervalSeconds.toLong * 1000L
+
+  // Reuse a single Configuration; GCS credentials come from Workload Identity or
+  // GOOGLE_APPLICATION_CREDENTIALS, picked up automatically by the GCS Hadoop connector.
+  private val conf = withClassLoader(new Configuration())
+
+  // Parquet schema: all data columns including audit fields (no partitioning).
+  private val parquetSchema: MessageType = new MessageType("access_log",
+    PTypes.required(BINARY).as(LogicalTypeAnnotation.stringType()).named("logType"),
+    PTypes.required(BINARY).as(LogicalTypeAnnotation.stringType()).named("share"),
+    PTypes.required(BINARY).as(LogicalTypeAnnotation.stringType()).named("schema"),
+    PTypes.required(BINARY).as(LogicalTypeAnnotation.stringType()).named("table"),
+    PTypes.required(INT64).named("egressBytes"),
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("pricingTier"),
+    PTypes.required(INT64).named("timestampMs"),
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("requestType"),
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("clientRegion"),
+    // Audit fields for customer audits and consolidated storage
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("tenantId"),
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("clientIp"),
+    PTypes.optional(BINARY).as(LogicalTypeAnnotation.stringType()).named("rawRegionHeader"),
+    PTypes.optional(BOOLEAN).named("isGcpIp")
+  )
+
+  // How often the flush loop wakes to check the stopped flag and time-based flush.
+  // Short enough for responsive shutdown; long enough to avoid busy-waiting.
+  private val CheckIntervalMs = 500L
+
+  private val flushThread: Thread = {
+    val t = new Thread(() => runFlushLoop(), "delta-access-log-writer")
+    t.setDaemon(true)
+    t.start()
+    t
+  }
+
+  override def record(entry: AccessLogEntry): Unit = {
+    if (entry.egressBytes <= 0) return
+    if (!queue.offer(entry)) {
+      logger.warn(
+        "Delta access log queue is full ({} capacity); " +
+          "dropping record for tenant {} share {}/{}/{}",
+        MaxQueueCapacity.asInstanceOf[AnyRef],
+        extractTenantId(entry.share),
+        entry.share,
+        entry.schema,
+        entry.table
+      )
+    }
+  }
+
+  override def recordContext(entry: PricingContextLogEntry): Unit = {}
+  override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {}
+
+  override def close(): Unit = {
+    stopped.set(true)
+    try {
+      flushThread.join(30000L)
+    } catch {
+      case _: InterruptedException =>
+        Thread.currentThread().interrupt()
+    }
+  }
+
+  private def runFlushLoop(): Unit = {
+    val batch = new java.util.ArrayList[AccessLogEntry]()
+    var lastFlushMs = System.currentTimeMillis()
+
+    while (!stopped.get()) {
+      try {
+        val head = queue.poll(CheckIntervalMs, TimeUnit.MILLISECONDS)
+        if (head != null) {
+          batch.add(head)
+          queue.drainTo(batch, flushBatchSize - 1)
+        }
+
+        val elapsed = System.currentTimeMillis() - lastFlushMs
+        val timeToFlush = elapsed >= flushIntervalMs
+        val batchFull = batch.size() >= flushBatchSize
+        if ((timeToFlush || batchFull) && batch.size() > 0) {
+          safeWriteBatch(batch.asScala.toList)
+          batch.clear()
+          lastFlushMs = System.currentTimeMillis()
+        }
+      } catch {
+        case _: InterruptedException => // ignore; re-check stopped on next iteration
+        case e: Exception =>
+          logger.error("Unexpected error in delta access log flush loop", e)
+      }
+    }
+
+    // Final flush: drain any remaining records before shutdown.
+    // Respect batch size to maintain consistent behavior (important for tests and
+    // scenarios where each record should produce a separate file).
+    queue.drainTo(batch)
+    if (!batch.isEmpty) {
+      val entries = batch.asScala.toList
+      entries.grouped(flushBatchSize).foreach(safeWriteBatch)
+    }
+  }
+
+  private def safeWriteBatch(entries: List[AccessLogEntry]): Unit = {
+    try {
+      withClassLoader(writeBatch(entries))
+    } catch {
+      case e: Exception =>
+        logger.error(s"Failed to write ${entries.size} access log entries to Delta table", e)
+    }
+  }
+
+  /**
+   * Extracts tenant_id from share name following the pattern `{tenant_id}_share`.
+   * Falls back to the share name itself if the pattern doesn't match.
+   */
+  private def extractTenantId(shareName: String): String = {
+    if (shareName.endsWith("_share")) {
+      shareName.dropRight("_share".length)
+    } else {
+      // Fallback for shares not following the convention
+      shareName
+    }
+  }
+
+  /**
+   * Returns the path to the consolidated access log table.
+   * Table is named access_log_br__system (double underscore because _system tenant starts with _)
+   */
+  private def consolidatedTablePath: String = {
+    val normalizedBase = if (basePath.endsWith("/")) basePath.dropRight(1) else basePath
+    s"$normalizedBase/access_log_br__system"
+  }
+
+  private def writeBatch(entries: List[AccessLogEntry]): Unit = {
+    // Write all entries to the single consolidated table
+    val tablePath = consolidatedTablePath
+
+    val deltaLog = DeltaLog.forTable(conf, new Path(tablePath))
+    val txn = deltaLog.startTransaction()
+
+    // Table must be pre-created by deltalake-admin tool
+    if (txn.readVersion() < 0) {
+      logger.error(
+        s"Delta table does not exist at $tablePath. " +
+        "Table must be created by deltalake-admin during tenant onboarding.")
+      return
+    }
+
+    // Enrich entries with tenantId if not already set
+    val enrichedEntries = entries.map { e =>
+      if (e.tenantId.isEmpty) {
+        e.copy(tenantId = Some(extractTenantId(e.share)))
+      } else {
+        e
+      }
+    }
+
+    val addFile: Option[io.delta.standalone.actions.Action] =
+      writeParquetFile(tablePath, enrichedEntries)
+
+    val addFiles: Seq[io.delta.standalone.actions.Action] = addFile.toSeq
+
+    if (addFiles.nonEmpty) {
+      val operation = new Operation(Operation.Name.WRITE)
+      txn.commit(addFiles.asJava, operation, "delta-sharing-server")
+    }
+  }
+
+  private def writeParquetFile(
+      tablePath: String,
+      entries: List[AccessLogEntry]): Option[io.delta.standalone.actions.Action] = {
+    val relPath = s"${UUID.randomUUID()}.parquet"
+    val absPath = new Path(s"$tablePath/$relPath")
+    try {
+      val factory = new SimpleGroupFactory(parquetSchema)
+      val parquetConf = new Configuration(conf)
+      val writer = ExampleParquetWriter.builder(absPath)
+        .withType(parquetSchema)
+        .withConf(parquetConf)
+        .withWriteMode(ParquetFileWriter.Mode.CREATE)
+        .withCompressionCodec(CompressionCodecName.SNAPPY)
+        .build()
+      try {
+        for (e <- entries) {
+          val g = factory.newGroup()
+          g.add("logType", Binary.fromString("ACCESS_LOG"))
+          g.add("share", Binary.fromString(e.share))
+          g.add("schema", Binary.fromString(e.schema))
+          g.add("table", Binary.fromString(e.table))
+          g.add("egressBytes", e.egressBytes)
+          g.add("pricingTier", Binary.fromString(e.pricingTier))
+          g.add("timestampMs", e.timestampMs)
+          g.add("requestType", Binary.fromString(e.requestType))
+          e.clientRegion.foreach(r => g.add("clientRegion", Binary.fromString(r)))
+          // Audit fields
+          e.tenantId.foreach(t => g.add("tenantId", Binary.fromString(t)))
+          e.clientIp.foreach(ip => g.add("clientIp", Binary.fromString(ip)))
+          e.rawRegionHeader.foreach(h => g.add("rawRegionHeader", Binary.fromString(h)))
+          e.isGcpIp.foreach(b => g.add("isGcpIp", b))
+          writer.write(g)
+        }
+      } finally {
+        writer.close()
+      }
+
+      val fs = absPath.getFileSystem(conf)
+      val fileSize = fs.getFileStatus(absPath).getLen
+      // No partition values - empty map
+      Some(AddFile.builder(relPath, java.util.Collections.emptyMap[String, String](),
+        fileSize, System.currentTimeMillis(), true).build())
+    } catch {
+      case e: Exception =>
+        logger.error(s"Failed to write parquet file $relPath", e)
+        None
+    }
+  }
+
+  private def withClassLoader[T](func: => T): T = {
+    val classLoader = Thread.currentThread().getContextClassLoader
+    if (classLoader == null) {
+      Thread.currentThread().setContextClassLoader(this.getClass.getClassLoader)
+      try func finally Thread.currentThread().setContextClassLoader(null)
+    } else {
+      func
+    }
+  }
+}
diff --git a/server/src/main/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookup.scala b/server/src/main/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookup.scala
index 780870d59..9cf6e6032 100644
--- a/server/src/main/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookup.scala
+++ b/server/src/main/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookup.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/server/src/main/scala/io/delta/sharing/server/telemetry/GcpPricingTier.scala b/server/src/main/scala/io/delta/sharing/server/telemetry/GcpPricingTier.scala
index 1fa1aada5..04eb88b4d 100644
--- a/server/src/main/scala/io/delta/sharing/server/telemetry/GcpPricingTier.scala
+++ b/server/src/main/scala/io/delta/sharing/server/telemetry/GcpPricingTier.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/server/src/test/scala/io/delta/sharing/server/telemetry/AccessLogEmitterSuite.scala b/server/src/test/scala/io/delta/sharing/server/telemetry/AccessLogEmitterSuite.scala
index da2762746..770870613 100644
--- a/server/src/test/scala/io/delta/sharing/server/telemetry/AccessLogEmitterSuite.scala
+++ b/server/src/test/scala/io/delta/sharing/server/telemetry/AccessLogEmitterSuite.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -106,6 +106,46 @@ class AccessLogEmitterSuite extends FunSuite {
     assert(emitter.isInstanceOf[JsonAccessLogEmitter])
   }
 
+  test("AccessLogEmitter.create returns CompositeAccessLogEmitter when deltaTablePath is set") {
+    val config = new ServerConfig()
+    val accessConfig = new AccessLoggingConfig()
+    accessConfig.setEnabled(true)
+    accessConfig.setDeltaTablePath("/tmp/test-delta-table")
+    config.setAccessLogging(accessConfig)
+
+    val emitter = AccessLogEmitter.create(config)
+    try {
+      assert(emitter.isInstanceOf[CompositeAccessLogEmitter])
+    } finally {
+      emitter.close()
+    }
+  }
+
+  test("CompositeAccessLogEmitter fans out record calls to all delegates") {
+    var count = 0
+    val counting = new AccessLogEmitter {
+      override def record(entry: AccessLogEntry): Unit = count += 1
+      override def recordContext(entry: PricingContextLogEntry): Unit = {}
+      override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {}
+    }
+    val composite = new CompositeAccessLogEmitter(Seq(counting, counting))
+    composite.record(AccessLogEntry("s", "sc", "t", 100L, 0L))
+    assert(count == 2)
+  }
+
+  test("CompositeAccessLogEmitter close() calls close on all delegates") {
+    var closedCount = 0
+    val closeable = new AccessLogEmitter {
+      override def record(entry: AccessLogEntry): Unit = {}
+      override def recordContext(entry: PricingContextLogEntry): Unit = {}
+      override def recordHeaders(entry: RequestHeadersLogEntry): Unit = {}
+      override def close(): Unit = closedCount += 1
+    }
+    val composite = new CompositeAccessLogEmitter(Seq(closeable, closeable))
+    composite.close()
+    assert(closedCount == 2)
+  }
+
   test("request type constants are defined") {
     assert(AccessLogEmitter.QueryRequestType == "query")
     assert(AccessLogEmitter.CdfStreamRequestType == "cdf_stream")
diff --git a/server/src/test/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriterSuite.scala b/server/src/test/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriterSuite.scala
new file mode 100644
index 000000000..c40ead064
--- /dev/null
+++ b/server/src/test/scala/io/delta/sharing/server/telemetry/DeltaAccessLogWriterSuite.scala
@@ -0,0 +1,293 @@
+/*
+ * Copyright (2026) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.sharing.server.telemetry
+
+import java.nio.file.Files
+
+import scala.collection.JavaConverters._
+
+import io.delta.standalone.DeltaLog
+import io.delta.standalone.Operation
+import io.delta.standalone.actions.{Format, Metadata, Protocol}
+import io.delta.standalone.types.{BooleanType, LongType, StringType, StructType}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.scalatest.FunSuite
+
+class DeltaAccessLogWriterSuite extends FunSuite {
+
+  private val conf = new Configuration()
+
+  // Delta schema matching what deltalake-admin would create
+  private val deltaSchema: StructType = new StructType()
+    .add("logType", new StringType(), false)
+    .add("share", new StringType(), false)
+    .add("schema", new StringType(), false)
+    .add("table", new StringType(), false)
+    .add("egressBytes", new LongType(), false)
+    .add("pricingTier", new StringType(), true)
+    .add("timestampMs", new LongType(), false)
+    .add("requestType", new StringType(), true)
+    .add("clientRegion", new StringType(), true)
+    .add("tenantId", new StringType(), true)
+    .add("clientIp", new StringType(), true)
+    .add("rawRegionHeader", new StringType(), true)
+    .add("isGcpIp", new BooleanType(), true)
+
+  private def makeTempPath(): String = {
+    val dir = Files.createTempDirectory("delta-access-log-test")
+    dir.toAbsolutePath.toString
+  }
+
+  private def makeEntry(
+      share: String = "tenant1_share",
+      schema: String = "sc",
+      table: String = "t",
+      egressBytes: Long = 1024L,
+      timestampMs: Long = 1717502400000L, // 2024-06-04 in UTC
+      pricingTier: String = "internet_to_na_eu",
+      requestType: String = "query",
+      clientRegion: Option[String] = Some("US"),
+      tenantId: Option[String] = None,
+      clientIp: Option[String] = Some("203.0.113.45")): AccessLogEntry =
+    AccessLogEntry(share, schema, table, egressBytes, timestampMs,
+      pricingTier, clientRegion, requestType, tenantId, clientIp,
+      rawRegionHeader = Some("US"), isGcpIp = Some(false))
+
+  /**
+   * Gets the consolidated table path from the base path.
+   * All records are written to access_log_br__system.
+   */
+  private def consolidatedTablePath(basePath: String): String = {
+    s"$basePath/access_log_br__system"
+  }
+
+  /**
+   * Pre-creates the Delta table with the expected schema.
+   * This simulates what deltalake-admin does during tenant onboarding.
+   */
+  private def createTable(basePath: String): Unit = {
+    val tablePath = consolidatedTablePath(basePath)
+    val deltaLog = DeltaLog.forTable(conf, new Path(tablePath))
+    val txn = deltaLog.startTransaction()
+
+    val metadata = Metadata.builder()
+      .schema(deltaSchema)
+      .format(new Format())
+      .partitionColumns(java.util.Collections.emptyList[String]())
+      .configuration(java.util.Collections.emptyMap[String, String]())
+      .createdTime(System.currentTimeMillis())
+      .build()
+    txn.updateMetadata(metadata)
+
+    val actions = Seq[io.delta.standalone.actions.Action](new Protocol(1, 2))
+    txn.commit(actions.asJava, new Operation(Operation.Name.CREATE_TABLE), "test")
+  }
+
+  test("single record is written to pre-created Delta table") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    writer.record(makeEntry())
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    val snapshot = log.snapshot()
+    assert(snapshot.getVersion >= 1, "Delta table should have commits after write")
+
+    val files = snapshot.getAllFiles.asScala
+    assert(files.nonEmpty, "Snapshot should contain at least one data file")
+
+    // Verify no partitioning
+    val partFile = files.head
+    assert(partFile.getPartitionValues.isEmpty, "Table should not be partitioned")
+  }
+
+  test("records from different timestamps are written to a single unpartitioned table") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    // Three entries spanning two different days
+    writer.record(makeEntry(timestampMs = 1717502400000L)) // 2024-06-04
+    writer.record(makeEntry(timestampMs = 1717502400000L)) // 2024-06-04
+    writer.record(makeEntry(timestampMs = 1717588800000L)) // 2024-06-05
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    val files = log.snapshot().getAllFiles.asScala.toList
+    // All records go to a single file since there's no partitioning
+    assert(files.size == 1, "One file expected (no partitioning)")
+    assert(files.head.getPartitionValues.isEmpty, "Table should not be partitioned")
+  }
+
+  test("multiple flushes produce multiple Delta commits") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    // batchSize=1 means each record triggers its own flush.
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 300, flushBatchSize = 1)
+    writer.record(makeEntry(table = "t1", timestampMs = 1717502400000L)) // 2024-06-04
+    writer.record(makeEntry(table = "t2", timestampMs = 1717588800000L)) // 2024-06-05
+    writer.close() // waits for all pending flushes
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    assert(log.snapshot().getVersion >= 1, "Table should have commits after write")
+    val files = log.snapshot().getAllFiles.asScala.toList
+    // Each flush writes a separate file
+    assert(files.size == 2, "Both records should produce separate files")
+    assert(files.forall(_.getPartitionValues.isEmpty), "Table should not be partitioned")
+  }
+
+  test("records with zero egressBytes are silently skipped") {
+    val basePath = makeTempPath()
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    writer.record(makeEntry(egressBytes = 0L))
+    writer.record(makeEntry(egressBytes = -1L))
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(new Configuration(), new Path(tablePath))
+    assert(log.snapshot().getVersion < 0 ||
+      log.snapshot().getAllFiles.asScala.isEmpty,
+      "No data files should exist when all records are skipped")
+  }
+
+  test("recordContext and recordHeaders are no-ops") {
+    val basePath = makeTempPath()
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    // Should not throw or write anything
+    writer.recordContext(PricingContextLogEntry("s", "t", 0L))
+    writer.recordHeaders(RequestHeadersLogEntry("s", "t", 0L, Map.empty))
+    writer.close()
+  }
+
+  test("write failures do not propagate to callers") {
+    // Use an invalid path to force write failure
+    val writer = new DeltaAccessLogWriter(
+      "/nonexistent/readonly/path",
+      flushIntervalSeconds = 5,
+      flushBatchSize = 1)
+    // Should not throw
+    writer.record(makeEntry())
+    writer.close()
+  }
+
+  test("close flushes records that have not yet been written") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    // Very long flush interval so nothing writes until close()
+    val writer =
+      new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 3600, flushBatchSize = 10000)
+    writer.record(makeEntry())
+    writer.record(makeEntry())
+    writer.close() // should trigger final flush
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    val files = log.snapshot().getAllFiles.asScala
+    assert(files.nonEmpty, "Records should be flushed on close()")
+  }
+
+  test("Parquet files contain the expected columns including audit fields") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    writer.record(makeEntry())
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    val addFile = log.snapshot().getAllFiles.asScala.head
+    val filePath = new Path(s"$tablePath/${addFile.getPath}")
+    val reader = ParquetFileReader.open(HadoopInputFile.fromPath(filePath, conf))
+    try {
+      val schema = reader.getFileMetaData.getSchema
+      val fields = schema.getFields.asScala.map(_.getName).toSet
+      // Core fields
+      assert(fields.contains("logType"))
+      assert(fields.contains("share"))
+      assert(fields.contains("schema"))
+      assert(fields.contains("table"))
+      assert(fields.contains("egressBytes"))
+      assert(fields.contains("pricingTier"))
+      assert(fields.contains("timestampMs"))
+      assert(fields.contains("requestType"))
+      assert(fields.contains("clientRegion"))
+      // Audit fields
+      assert(fields.contains("tenantId"))
+      assert(fields.contains("clientIp"))
+      assert(fields.contains("rawRegionHeader"))
+      assert(fields.contains("isGcpIp"))
+    } finally {
+      reader.close()
+    }
+  }
+
+  test("records are NOT written when table does not exist") {
+    val basePath = makeTempPath()
+    // Do NOT create table - simulates missing deltalake-admin setup
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    writer.record(makeEntry())
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    // Table should still not exist - records are dropped
+    assert(log.snapshot().getVersion < 0, "Table should not be created by writer")
+  }
+
+  test("records from different tenants are written to the same consolidated table") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    writer.record(makeEntry(share = "tenant1_share"))
+    writer.record(makeEntry(share = "tenant2_share"))
+    writer.record(makeEntry(share = "_system_share"))
+    writer.close()
+
+    // Check consolidated table has data
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    assert(log.snapshot().getVersion >= 1, "Consolidated table should have commits")
+    assert(log.snapshot().getAllFiles.asScala.nonEmpty, "Consolidated table should have data")
+  }
+
+  test("tenantId is auto-derived from share name when not provided") {
+    val basePath = makeTempPath()
+    createTable(basePath)
+
+    val writer = new DeltaAccessLogWriter(basePath, flushIntervalSeconds = 5, flushBatchSize = 100)
+    // Record without explicit tenantId
+    writer.record(makeEntry(share = "my_tenant_share", tenantId = None))
+    writer.close()
+
+    val tablePath = consolidatedTablePath(basePath)
+    val log = DeltaLog.forTable(conf, new Path(tablePath))
+    assert(log.snapshot().getVersion >= 1, "Table should have commits")
+  }
+}
diff --git a/server/src/test/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookupSuite.scala b/server/src/test/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookupSuite.scala
index 08ac384b3..541b5580b 100644
--- a/server/src/test/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookupSuite.scala
+++ b/server/src/test/scala/io/delta/sharing/server/telemetry/GcpIpRangeLookupSuite.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/server/src/test/scala/io/delta/sharing/server/telemetry/GcpPricingTierSuite.scala b/server/src/test/scala/io/delta/sharing/server/telemetry/GcpPricingTierSuite.scala
index 3317951d7..795090323 100644
--- a/server/src/test/scala/io/delta/sharing/server/telemetry/GcpPricingTierSuite.scala
+++ b/server/src/test/scala/io/delta/sharing/server/telemetry/GcpPricingTierSuite.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright (2021) The Delta Lake Project Authors.
+ * Copyright (2026) The Delta Lake Project Authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/server/src/universal/conf/delta-sharing-server.yaml.template b/server/src/universal/conf/delta-sharing-server.yaml.template
index fffd0a76a..29a005df3 100644
--- a/server/src/universal/conf/delta-sharing-server.yaml.template
+++ b/server/src/universal/conf/delta-sharing-server.yaml.template
@@ -88,3 +88,13 @@ accessLogging:
   # When true, client IPs belonging to other GCP regions can be classified as inter-region
   # (cheaper) rather than internet egress. Set to false to disable this detection.
   detectGcpTraffic: true
+  # GCS base path for the consolidated access log Delta table. When set, ACCESS_LOG entries are
+  # written asynchronously to `{deltaTablePath}/access_log_br__system` in addition to the JSON log stream.
+  # The Delta table must be pre-created by deltalake-admin; the server does not auto-create it.
+  # Omit or leave empty to disable Delta writing.
+  # Example: gs://my-bucket/datalake/data/tenant/_system
+  # deltaTablePath: ""
+  # How often (seconds) to flush buffered access log records to the Delta table.
+  # deltaFlushIntervalSeconds: 60
+  # Maximum number of records to buffer before triggering an early flush.
+  # deltaFlushBatchSize: 1000