Polaris Tag Management: tag assignment write path

CHANGELOG.md

@@ -32,7 +32,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### Upgrade notes
 
 - Relational JDBC: schema version 6 corrects the `idx_locations` index on Postgres and CockroachDB
-  (see Fixes). Fresh bootstraps use schema v6 automatically and get the right index. Because Polaris
+  (see Fixes). Fresh bootstraps use the latest schema version (v7, see the schema v7 upgrade note
+  below) and get the right index. Because Polaris
   has no automated schema migrations, existing Postgres/CockroachDB deployments keep the old,
   ineffective index until an operator recreates it manually:
   ```sql
@@ -54,12 +55,60 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
   ```
   Postgres and H2 are unaffected.
 
+- Relational JDBC: a new schema version 7 adds the `tag_assignment_record` table backing tag
+  assignments. Fresh bootstraps use schema v7 automatically. Existing deployments on older schema
+  versions keep working, but tag assignment writes are rejected with an error naming the v7
+  requirement until the database is upgraded; Polaris has no automated schema migrations, so the
+  upgrade is a one-time manual step (run against each existing database, then restart Polaris).
+  A database below schema v5 must first apply the v5 change (see the 1.7.0 release's Upgrade
+  notes), then the v6 index changes above, then this step, in that order. Skipping the v5 step
+  and jumping straight from schema v3/v4 to v7 leaves the `events.catalog_id` column `NOT NULL`
+  while the server, now reporting schema v7, writes `NULL` there, so every event write that is not
+  catalog-scoped fails. New realms bootstrapped into an existing database stay on that database's
+  current schema version, so this step is required before tag assignments can be used anywhere on
+  it:
+  ```sql
+  CREATE TABLE IF NOT EXISTS polaris_schema.tag_assignment_record (
+      realm_id TEXT NOT NULL,
+      target_catalog_id BIGINT NOT NULL,
+      target_id BIGINT NOT NULL,
+      field_id INTEGER NOT NULL DEFAULT 0,
+      tag_catalog_id BIGINT NOT NULL,
+      tag_id BIGINT NOT NULL,
+      tag_value TEXT NOT NULL,
+      PRIMARY KEY (realm_id, target_catalog_id, target_id, field_id, tag_catalog_id, tag_id)
+  );
+  CREATE INDEX IF NOT EXISTS idx_tag_assignment_record_by_tag
+      ON polaris_schema.tag_assignment_record (realm_id, tag_catalog_id, tag_id);
+  CREATE INDEX IF NOT EXISTS idx_tag_assignment_record_by_tag_value
+      ON polaris_schema.tag_assignment_record (realm_id, tag_catalog_id, tag_id, tag_value);
+  UPDATE polaris_schema.version SET version_value = 7 WHERE version_key = 'version';
+  ```
+  On CockroachDB, declare `field_id INT4` instead of `INTEGER`, matching the shipped
+  `cockroachdb/schema-v7.sql` (the generic INTEGER type maps incorrectly in the CockroachDB
+  JDBC driver).
+
 ### Breaking changes
 
 - Concurrent table commits that hit a stale sequence number now return a retryable `409` instead of a fatal `400`, for both single-table commits and `commitTransaction`.
 
 ### New Features
 
+- Added tag management: tag definitions can be created, listed, loaded, updated and dropped
+  through the new `/polaris/v1/{prefix}/tags` endpoints, with catalog-scoped authorization and
+  new `TAG_*` privileges covered by `CATALOG_MANAGE_CONTENT`. The feature is gated by the
+  `ENABLE_TAG_STORE` feature flag (disabled by default) and is supported on the JDBC and
+  in-memory metastores; the NoSQL metastore does not support tags yet. Tag assignments are
+  included: tags can be assigned to and unassigned from catalogs, namespaces, tables and
+  top-level Iceberg table columns through the `/tags/{tag}/mappings` endpoints, and
+  `detach-all=true` on drop removes a definition together with its assignments; no Tag API
+  read observes the definition without its assignments or a partially removed state; a backend
+  that cannot guarantee that result answers 501 and changes nothing. Allowed
+  values and assigned tag values are limited to 2000 UTF-8 bytes, an implementation size limit:
+  the value is part of a database index key. An assignment write checks the selected value
+  against the tag definition inside its own transaction and conflicts with a concurrent
+  allowed-values update on that same definition row, so a value an update removes before the
+  assignment commits is rejected.
 - Python CLI: `catalogs update` now supports `--no-sts` and `--no-kms` to toggle STS/KMS availability on an existing S3 catalog. Previously these were only settable at `catalogs create` time.
 - Python CLI: added `gcp` as an external catalog authentication type for Iceberg REST federation, enabling CLI creation of GCP-authenticated catalogs such as BigLake without passing Google credential secrets through command-line flags.
 

api/polaris-catalog-service/build.gradle.kts

@@ -49,7 +49,25 @@ val policyManagementModels =
     "ListPoliciesResponse",
   )
 
-val models = (genericTableModels + policyManagementModels).joinToString(",")
+val tagManagementModels =
+  listOf(
+    "TargetType",
+    "Tag",
+    "TagIdentifier",
+    "CreateTagRequest",
+    "UpdateTagRequest",
+    "LoadTagResponse",
+    "ListTagsResponse",
+    "TagAttachmentTarget",
+    "AssignTagRequest",
+    "UnassignTagRequest",
+    "ObjectTag",
+    "GetObjectTagsResponse",
+    "TaggedObject",
+    "ListObjectsByTagResponse",
+  )
+
+val models = (genericTableModels + policyManagementModels + tagManagementModels).joinToString(",")
 
 dependencies {
   implementation(project(":polaris-core"))
@@ -101,7 +119,7 @@ openApiGenerate {
   ignoreFileOverride.set(provider { rootDir.file(".openapi-generator-ignore").asFile.absolutePath })
   removeOperationIdPrefix.set(true)
   templateDir.set(provider { templatesDir.asFile.absolutePath })
-  globalProperties.put("apis", "GenericTableApi,PolicyApi")
+  globalProperties.put("apis", "GenericTableApi,PolicyApi,TagApi")
   globalProperties.put("models", models)
   globalProperties.put("apiDocs", "false")
   globalProperties.put("modelTests", "false")

extensions/auth/opa/src/main/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizer.java

@@ -54,6 +54,7 @@
 import org.apache.polaris.core.auth.RoleAssignmentAuthorizationIntent;
 import org.apache.polaris.core.auth.RootPrivilegeGrantAuthorizationIntent;
 import org.apache.polaris.core.auth.SingleTargetAuthorizationIntent;
+import org.apache.polaris.core.auth.TagAttachmentAuthorizationIntent;
 import org.apache.polaris.core.auth.TargetlessAuthorizationIntent;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
@@ -160,6 +161,10 @@ public AuthorizationDecision authorize(
           targets = toResourceEntitiesFromSecurable(policyAttachmentIntent.policy());
           secondaries = toResourceEntitiesFromSecurable(policyAttachmentIntent.attachedTo());
         }
+        case TagAttachmentAuthorizationIntent tagAttachmentIntent -> {
+          targets = toResourceEntitiesFromSecurable(tagAttachmentIntent.tag());
+          secondaries = toResourceEntitiesFromSecurable(tagAttachmentIntent.attachedTo());
+        }
         case RoleAssignmentAuthorizationIntent roleAssignmentIntent -> {
           targets = toResourceEntitiesFromSecurable(roleAssignmentIntent.role());
           secondaries = toResourceEntitiesFromSecurable(roleAssignmentIntent.assignee());

extensions/auth/opa/src/test/java/org/apache/polaris/extension/auth/opa/OpaPolarisAuthorizerTest.java

@@ -59,6 +59,7 @@
 import org.apache.polaris.core.auth.PolarisSecurable;
 import org.apache.polaris.core.auth.RenameAuthorizationIntent;
 import org.apache.polaris.core.auth.SingleTargetAuthorizationIntent;
+import org.apache.polaris.core.auth.TagAttachmentAuthorizationIntent;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -990,6 +991,126 @@ <T> T httpClientExecute(
         .isEqualTo(expectedDestination);
   }
 
+  @Test
+  void authorizeTagAttachmentIncludesTagTargetAndAttachedToSecondary() throws Exception {
+    final String[] capturedRequestBody = new String[1];
+    HttpEntity mockEntity = HttpEntities.create("{\"result\":{\"allow\":true}}");
+    @SuppressWarnings("resource")
+    ClassicHttpResponse mockResponse = new BasicClassicHttpResponse(200);
+    mockResponse.setEntity(mockEntity);
+
+    PolarisResolutionManifest resolutionManifest = mock(PolarisResolutionManifest.class);
+    AuthorizationState authzState = new AuthorizationState(resolutionManifest);
+
+    AuthorizationRequest request =
+        new AuthorizationRequest(
+            PolarisPrincipal.of("alice", Map.of(), Set.of("role-1")),
+            List.of(
+                new TagAttachmentAuthorizationIntent(
+                    PolarisAuthorizableOperation.ASSIGN_TAG_TO_TABLE,
+                    PolarisSecurable.of(
+                        new PathSegment(PolarisEntityType.CATALOG, "catalog1"),
+                        new PathSegment(PolarisEntityType.TAG, "pii")),
+                    PolarisSecurable.of(
+                        new PathSegment(PolarisEntityType.CATALOG, "catalog1"),
+                        new PathSegment(PolarisEntityType.NAMESPACE, "ns"),
+                        new PathSegment(PolarisEntityType.TABLE_LIKE, "tbl")))));
+
+    OpaPolarisAuthorizer authorizer =
+        new OpaPolarisAuthorizer(
+            URI.create("http://opa.example.com:8181/v1/data/polaris/allow"),
+            mock(CloseableHttpClient.class),
+            JsonMapper.builder().build(),
+            null,
+            null,
+            "test-realm") {
+          @Override
+          <T> T httpClientExecute(
+              ClassicHttpRequest request, HttpClientResponseHandler<? extends T> responseHandler)
+              throws HttpException, IOException {
+            capturedRequestBody[0] =
+                new String(request.getEntity().getContent().readAllBytes(), StandardCharsets.UTF_8);
+            return responseHandler.handleResponse(mockResponse);
+          }
+        };
+
+    AuthorizationDecision decision = authorizer.authorize(authzState, request);
+    ObjectMapper mapper = JsonMapper.builder().build();
+    JsonNode root = mapper.readTree(capturedRequestBody[0]);
+
+    assertThat(decision.isAllowed()).isTrue();
+
+    JsonNode expectedTag =
+        mapper.readTree(
+            """
+            {
+              "type": "TAG",
+              "name": "pii",
+              "parents": [
+                {"type": "CATALOG", "name": "catalog1", "parents": []}
+              ]
+            }
+            """);
+    assertThat(root.path("input").path("resource").path("targets").get(0)).isEqualTo(expectedTag);
+
+    JsonNode expectedAttachedTo =
+        mapper.readTree(
+            """
+            {
+              "type": "TABLE_LIKE",
+              "name": "tbl",
+              "parents": [
+                {"type": "CATALOG", "name": "catalog1", "parents": []},
+                {"type": "NAMESPACE", "name": "ns", "parents": []}
+              ]
+            }
+            """);
+    assertThat(root.path("input").path("resource").path("secondaries").get(0))
+        .isEqualTo(expectedAttachedTo);
+  }
+
+  @Test
+  void authorizeTagAttachmentDeniedByPolicy() throws Exception {
+    HttpEntity mockEntity = HttpEntities.create("{\"result\":{\"allow\":false}}");
+    @SuppressWarnings("resource")
+    ClassicHttpResponse mockResponse = new BasicClassicHttpResponse(200);
+    mockResponse.setEntity(mockEntity);
+
+    PolarisResolutionManifest resolutionManifest = mock(PolarisResolutionManifest.class);
+    AuthorizationState authzState = new AuthorizationState(resolutionManifest);
+
+    AuthorizationRequest request =
+        new AuthorizationRequest(
+            PolarisPrincipal.of("alice", Map.of(), Set.of("role-1")),
+            List.of(
+                new TagAttachmentAuthorizationIntent(
+                    PolarisAuthorizableOperation.UNASSIGN_TAG_FROM_NAMESPACE,
+                    PolarisSecurable.of(
+                        new PathSegment(PolarisEntityType.CATALOG, "catalog1"),
+                        new PathSegment(PolarisEntityType.TAG, "pii")),
+                    PolarisSecurable.of(
+                        new PathSegment(PolarisEntityType.CATALOG, "catalog1"),
+                        new PathSegment(PolarisEntityType.NAMESPACE, "ns")))));
+
+    OpaPolarisAuthorizer authorizer =
+        new OpaPolarisAuthorizer(
+            URI.create("http://opa.example.com:8181/v1/data/polaris/allow"),
+            mock(CloseableHttpClient.class),
+            JsonMapper.builder().build(),
+            null,
+            null,
+            "test-realm") {
+          @Override
+          <T> T httpClientExecute(
+              ClassicHttpRequest request, HttpClientResponseHandler<? extends T> responseHandler)
+              throws HttpException, IOException {
+            return responseHandler.handleResponse(mockResponse);
+          }
+        };
+
+    assertThat(authorizer.authorize(authzState, request).isAllowed()).isFalse();
+  }
+
   @Test
   void authorizeSingleOperationMultiIntentRequestEvaluatesSequentially() throws Exception {
     final List<String> capturedRequestBodies = new ArrayList<>();

integration-tests/src/main/java/org/apache/polaris/service/it/env/PolarisClient.java

@@ -127,6 +127,10 @@ public PolicyApi policyApi(String authToken) {
     return new PolicyApi(client, endpoints, authToken, endpoints.catalogApiEndpoint());
   }
 
+  public TagApi tagApi(String authToken) {
+    return new TagApi(client, endpoints, authToken, endpoints.catalogApiEndpoint());
+  }
+
   public PlatformMetricsApi platformMetricsApi() {
     if (platformEndpoints == null) {
       throw new IllegalStateException("Platform endpoints are not available");
@@ -178,6 +182,9 @@ public void cleanUp(String authToken) {
         .forEach(
             c -> {
               catalogApi.purge(c.getName());
+              // a live tag definition blocks the catalog drop, so purge tags too; tolerate
+              // only 406 (feature disabled)
+              tagApi(authToken).purgeIfAvailable(c.getName());
               managementApi.dropCatalog(c.getName());
             });