feat(persistence): replace EventEntity.REALM_SCOPED sentinel with nullable catalog_id

CHANGELOG.md

@@ -30,6 +30,18 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### Highlights
 
 ### Upgrade notes
+- Relational JDBC: a new schema version 5 makes the `events.catalog_id` column nullable. Events that
+  are not scoped to a catalog (principal, policy, rate-limiting, etc.) now persist `NULL` instead of
+  the placeholder string `__realm__` (which only ever existed in 1.6.0 release candidates). Fresh
+  bootstraps use schema v5 automatically. Existing deployments on schema v3/v4 keep working — the
+  server detects the older schema version and continues writing the legacy placeholder — but
+  operators are encouraged to upgrade manually, since Polaris has no automated schema migrations:
+  ```sql
+  ALTER TABLE polaris_schema.events ALTER COLUMN catalog_id DROP NOT NULL;
+  UPDATE polaris_schema.events SET catalog_id = NULL WHERE catalog_id = '__realm__';
+  UPDATE polaris_schema.version SET version_value = 5 WHERE version_key = 'version';
+  ```
+  See the Relational JDBC metastore documentation for details.
 
 ### Breaking changes
 - Removed the `--schema-version` (`-v`) option from the admin tool's `bootstrap` command. New realms

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -286,7 +286,7 @@ public void writeEvents(@NonNull List<EventEntity> events) {
           QueryGenerator.generateInsertQuery(
               ModelEvent.ALL_COLUMNS,
               ModelEvent.TABLE_NAME,
-              ModelEvent.fromEvent(events.getFirst())
+              ModelEvent.fromEvent(events.getFirst(), schemaVersion)
                   .toMap(datasourceOperations.getDatabaseType())
                   .values()
                   .stream()
@@ -304,7 +304,7 @@ public void writeEvents(@NonNull List<EventEntity> events) {
             QueryGenerator.generateInsertQuery(
                 ModelEvent.ALL_COLUMNS,
                 ModelEvent.TABLE_NAME,
-                ModelEvent.fromEvent(event)
+                ModelEvent.fromEvent(event, schemaVersion)
                     .toMap(datasourceOperations.getDatabaseType())
                     .values()
                     .stream()

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/models/ModelEvent.java

@@ -43,6 +43,18 @@ public interface ModelEvent extends Converter<EventEntity> {
   String RESOURCE_IDENTIFIER = "resource_identifier";
   String ADDITIONAL_PROPERTIES = "additional_properties";
 
+  /**
+   * Legacy sentinel stored in {@code events.catalog_id} for events that are not catalog-scoped on
+   * schema versions &lt; 5, where the column is {@code NOT NULL}. Never exposed outside the JDBC
+   * layer: {@link #fromEvent(EventEntity, int)} substitutes it for a null catalog id on pre-v5
+   * schemas, and {@link #fromResultSet} coalesces it back to null on read.
+   *
+   * <p>Because {@link #fromResultSet} cannot know which schema version a row was written under, the
+   * coalescing is unconditional, so this string remains reserved on all schema versions: a catalog
+   * literally named {@code __realm__} would have its events read back as realm-scoped.
+   */
+  String LEGACY_REALM_SCOPED_CATALOG_ID = "__realm__";
+
   List<String> ALL_COLUMNS =
       List.of(
           CATALOG_ID,
@@ -62,7 +74,6 @@ public interface ModelEvent extends Converter<EventEntity> {
    */
   ModelEvent CONVERTER =
       ImmutableModelEvent.builder()
-          .catalogId("")
           .eventId("")
           .requestId("")
           .eventType("")
@@ -73,8 +84,8 @@ public interface ModelEvent extends Converter<EventEntity> {
           .additionalProperties("")
           .build();
 
-  // catalog id
-  String getCatalogId();
+  // catalog id, null for events that are not catalog-scoped
+  @Nullable String getCatalogId();
 
   // event id
   String getEventId();
@@ -102,9 +113,10 @@ public interface ModelEvent extends Converter<EventEntity> {
 
   @Override
   default EventEntity fromResultSet(ResultSet rs) throws SQLException {
+    String catalogId = rs.getString(CATALOG_ID);
     var modelEvent =
         ImmutableModelEvent.builder()
-            .catalogId(rs.getString(CATALOG_ID))
+            .catalogId(LEGACY_REALM_SCOPED_CATALOG_ID.equals(catalogId) ? null : catalogId)
             .eventId(rs.getString(EVENT_ID))
             .requestId(rs.getString(REQUEST_ID))
             .eventType(rs.getString(EVENT_TYPE))
@@ -136,11 +148,16 @@ default Map<String, Object> toMap(DatabaseType databaseType) {
     return map;
   }
 
-  static ModelEvent fromEvent(EventEntity event) {
+  static ModelEvent fromEvent(EventEntity event, int schemaVersion) {
     if (event == null) return null;
 
+    String catalogId = event.getCatalogId();
+    if (schemaVersion < 5 && catalogId == null) {
+      // Schema versions < 5 declare events.catalog_id NOT NULL; store the legacy sentinel there.
+      catalogId = LEGACY_REALM_SCOPED_CATALOG_ID;
+    }
     return ImmutableModelEvent.builder()
-        .catalogId(event.getCatalogId())
+        .catalogId(catalogId)
         .eventId(event.getId())
         .requestId(event.getRequestId())
         .eventType(event.getEventType())

persistence/relational-jdbc/src/main/resources/cockroachdb/schema-v5.sql

@@ -21,6 +21,7 @@
 -- Changes from v4:
 --  * Removed the `idempotency_records` table (the durable idempotency store it
 --    backed was never wired into any request path and has been removed)
+--  * `events.catalog_id` is nullable; events that are not catalog-scoped store NULL (issue #4674)
 -- Features:
 --  * Uses INT4 explicitly for all integer columns (required for CockroachDB JDBC driver)
 --  * Includes all tables: version, entities, grant_records, principal_authentication_data,
@@ -131,7 +132,7 @@ CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (r
 
 CREATE TABLE IF NOT EXISTS events (
     realm_id TEXT NOT NULL,
-    catalog_id TEXT NOT NULL,
+    catalog_id TEXT,
     event_id TEXT NOT NULL,
     request_id TEXT,
     event_type TEXT NOT NULL,

persistence/relational-jdbc/src/main/resources/h2/schema-v5.sql

@@ -20,6 +20,7 @@
 -- Changes from v4:
 --  * Removed the `idempotency_records` table (the durable idempotency store it
 --    backed was never wired into any request path and has been removed)
+--  * `events.catalog_id` is nullable; events that are not catalog-scoped store NULL (issue #4674)
 
 CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
 SET SCHEMA POLARIS_SCHEMA;
@@ -130,7 +131,7 @@ CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (r
 
 CREATE TABLE IF NOT EXISTS events (
     realm_id TEXT NOT NULL,
-    catalog_id TEXT NOT NULL,
+    catalog_id TEXT,
     event_id TEXT NOT NULL,
     request_id TEXT,
     event_type TEXT NOT NULL,

persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql

@@ -19,6 +19,7 @@
 -- Changes from v4:
 --  * Removed the `idempotency_records` table (the durable idempotency store it
 --    backed was never wired into any request path and has been removed)
+--  * `events.catalog_id` is nullable; events that are not catalog-scoped store NULL (issue #4674)
 
 CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
 SET search_path TO POLARIS_SCHEMA;
@@ -130,7 +131,7 @@ CREATE INDEX IF NOT EXISTS idx_policy_mapping_record ON policy_mapping_record (r
 
 CREATE TABLE IF NOT EXISTS events (
     realm_id TEXT NOT NULL,
-    catalog_id TEXT NOT NULL,
+    catalog_id TEXT,
     event_id TEXT NOT NULL,
     request_id TEXT,
     event_type TEXT NOT NULL,

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java

@@ -110,7 +110,7 @@ void executeBatchUpdate_success() throws Exception {
           ImmutableModelEvent.builder()
               .resourceType(EventEntity.ResourceType.CATALOG)
               .resourceIdentifier("catalog_" + i)
-              .catalogId("catalog_" + i)
+              .catalogId(i % 2 == 0 ? "catalog_" + i : null)
               .eventId("event_" + i)
               .requestId("request_" + i)
               .eventType("event_type1")

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcEventsPersistenceTest.java

@@ -0,0 +1,119 @@
+/*
+ * 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.polaris.persistence.relational.jdbc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Stream;
+import javax.sql.DataSource;
+import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
+import org.apache.polaris.core.entity.EventEntity;
+import org.apache.polaris.core.persistence.PrincipalSecretsGenerator;
+import org.apache.polaris.persistence.relational.jdbc.models.ModelEvent;
+import org.h2.jdbcx.JdbcConnectionPool;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Tests event persistence across schema versions: schema versions before v5 declare
+ * events.catalog_id NOT NULL, so realm-scoped events (null catalog id) must be stored with the
+ * legacy sentinel; from v5 on they are stored as SQL NULL.
+ */
+class JdbcEventsPersistenceTest {
+
+  static Stream<Integer> schemaVersions() {
+    // The events table exists from schema v3 on.
+    return SchemaVersions.discoverAsStream(DatabaseType.H2).filter(version -> version >= 3);
+  }
+
+  @ParameterizedTest
+  @MethodSource("schemaVersions")
+  void writeEventsStoresNullCatalogIdPerSchemaVersion(int schemaVersion) throws Exception {
+    DataSource dataSource =
+        JdbcConnectionPool.create(
+            "jdbc:h2:mem:test_events_" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", "");
+    DatasourceOperations datasourceOperations =
+        new DatasourceOperations(
+            dataSource, SimpleRelationalJdbcConfiguration.forDatabaseType(DatabaseType.H2));
+    try (InputStream schemaStream = DatabaseType.H2.openInitScriptResource(schemaVersion)) {
+      datasourceOperations.executeScript(schemaStream);
+    }
+    JdbcBasePersistenceImpl persistence =
+        new JdbcBasePersistenceImpl(
+            new PolarisDefaultDiagServiceImpl(),
+            datasourceOperations,
+            PrincipalSecretsGenerator.RANDOM_SECRETS,
+            "TEST_REALM",
+            schemaVersion);
+
+    EventEntity realmScopedEvent =
+        new EventEntity(
+            null,
+            "realm-event",
+            null,
+            "CREATE_PRINCIPAL",
+            1234L,
+            "test-user",
+            EventEntity.ResourceType.REALM,
+            "principal-1");
+    EventEntity catalogScopedEvent =
+        new EventEntity(
+            "catalog-1",
+            "catalog-event",
+            "req-1",
+            "CREATE_TABLE",
+            1234L,
+            "test-user",
+            EventEntity.ResourceType.TABLE,
+            "table-1");
+
+    persistence.writeEvents(List.of(realmScopedEvent, catalogScopedEvent));
+
+    if (schemaVersion < 5) {
+      assertEquals(
+          ModelEvent.LEGACY_REALM_SCOPED_CATALOG_ID,
+          readStoredCatalogId(dataSource, "realm-event"));
+    } else {
+      assertNull(readStoredCatalogId(dataSource, "realm-event"));
+    }
+    assertEquals("catalog-1", readStoredCatalogId(dataSource, "catalog-event"));
+  }
+
+  private static String readStoredCatalogId(DataSource dataSource, String eventId)
+      throws Exception {
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement statement =
+            connection.prepareStatement(
+                "SELECT catalog_id FROM POLARIS_SCHEMA.EVENTS WHERE event_id = ?")) {
+      statement.setString(1, eventId);
+      try (ResultSet resultSet = statement.executeQuery()) {
+        assertTrue(resultSet.next());
+        return resultSet.getString(1);
+      }
+    }
+  }
+}

persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/models/ModelEventTest.java

@@ -30,6 +30,7 @@
 import static org.apache.polaris.persistence.relational.jdbc.models.ModelEvent.TIMESTAMP_MS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -54,6 +55,7 @@ public class ModelEventTest {
   private static final String TEST_RESOURCE_IDENTIFIER = "test-table";
   private static final String EMPTY_JSON = "{}";
   private static final String TEST_JSON = "{\"key\":\"value\"}";
+  private static final int LATEST_SCHEMA_VERSION = DatabaseType.H2.getLatestSchemaVersion();
 
   @Test
   public void testFromResultSet() throws SQLException {
@@ -84,6 +86,78 @@ public void testFromResultSet() throws SQLException {
     assertEquals(EMPTY_JSON, result.getAdditionalProperties());
   }
 
+  @Test
+  public void testFromResultSetCoalescesLegacyRealmScopedCatalogId() throws SQLException {
+    // Arrange
+    ResultSet mockResultSet = mock(ResultSet.class);
+    when(mockResultSet.getString(CATALOG_ID)).thenReturn(ModelEvent.LEGACY_REALM_SCOPED_CATALOG_ID);
+    when(mockResultSet.getString(EVENT_ID)).thenReturn(TEST_EVENT_ID);
+    when(mockResultSet.getString(REQUEST_ID)).thenReturn(TEST_REQUEST_ID);
+    when(mockResultSet.getString(EVENT_TYPE)).thenReturn(TEST_EVENT_TYPE);
+    when(mockResultSet.getLong(TIMESTAMP_MS)).thenReturn(TEST_TIMESTAMP_MS);
+    when(mockResultSet.getString(PRINCIPAL_NAME)).thenReturn(TEST_USER);
+    when(mockResultSet.getString(RESOURCE_TYPE)).thenReturn(TEST_RESOURCE_TYPE_STRING);
+    when(mockResultSet.getString(RESOURCE_IDENTIFIER)).thenReturn(TEST_RESOURCE_IDENTIFIER);
+    when(mockResultSet.getString(ADDITIONAL_PROPERTIES)).thenReturn(EMPTY_JSON);
+
+    // Act
+    EventEntity result = ModelEvent.CONVERTER.fromResultSet(mockResultSet);
+
+    // Assert
+    assertNull(result.getCatalogId());
+    assertEquals(TEST_EVENT_ID, result.getId());
+  }
+
+  @Test
+  public void testRoundTripWithNullCatalogId() {
+    // Arrange
+    EventEntity polarisEvent =
+        new EventEntity(
+            null,
+            TEST_EVENT_ID,
+            TEST_REQUEST_ID,
+            TEST_EVENT_TYPE,
+            TEST_TIMESTAMP_MS,
+            TEST_USER,
+            TEST_RESOURCE_TYPE,
+            TEST_RESOURCE_IDENTIFIER);
+    polarisEvent.setAdditionalProperties(TEST_JSON);
+
+    // Act
+    ModelEvent modelEvent = ModelEvent.fromEvent(polarisEvent, LATEST_SCHEMA_VERSION);
+    Map<String, Object> resultMap = modelEvent.toMap(DatabaseType.H2);
+    EventEntity result = ModelEvent.toEvent(modelEvent);
+
+    // Assert
+    assertNull(modelEvent.getCatalogId());
+    assertTrue(resultMap.containsKey(CATALOG_ID));
+    assertNull(resultMap.get(CATALOG_ID));
+    assertNull(result.getCatalogId());
+    assertEquals(TEST_EVENT_ID, result.getId());
+    assertEquals(TEST_JSON, result.getAdditionalProperties());
+  }
+
+  @Test
+  public void testFromEventStoresLegacySentinelForNullCatalogIdOnPreV5Schema() {
+    // Arrange
+    EventEntity polarisEvent =
+        new EventEntity(
+            null,
+            TEST_EVENT_ID,
+            TEST_REQUEST_ID,
+            TEST_EVENT_TYPE,
+            TEST_TIMESTAMP_MS,
+            TEST_USER,
+            TEST_RESOURCE_TYPE,
+            TEST_RESOURCE_IDENTIFIER);
+
+    // Act
+    ModelEvent modelEvent = ModelEvent.fromEvent(polarisEvent, 4);
+
+    // Assert
+    assertEquals(ModelEvent.LEGACY_REALM_SCOPED_CATALOG_ID, modelEvent.getCatalogId());
+  }
+
   @Test
   public void testToMapWithH2DatabaseType() {
     // Arrange
@@ -153,7 +227,7 @@ public void testToMapWithPostgresType() {
   @Test
   public void testFromEventWithNullInput() {
     // Act
-    ModelEvent result = ModelEvent.fromEvent(null);
+    ModelEvent result = ModelEvent.fromEvent(null, LATEST_SCHEMA_VERSION);
 
     // Assert
     assertNull(result);
@@ -175,7 +249,7 @@ public void testFromEvent() {
     polarisEvent.setAdditionalProperties(TEST_JSON);
 
     // Act
-    ModelEvent result = ModelEvent.fromEvent(polarisEvent);
+    ModelEvent result = ModelEvent.fromEvent(polarisEvent, LATEST_SCHEMA_VERSION);
 
     // Assert
     assertEquals(TEST_CATALOG_ID, result.getCatalogId());