This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new cf2988baa66 [feature](fe) Support iceberg REST catalog labels as the
$labels table (#69135)
cf2988baa66 is described below
commit cf2988baa66b0836a491d377622857e1cc36eb27
Author: Andrei Tserakhau <[email protected]>
AuthorDate: Tue Oct 27 10:00:05 2026 +0100
[feature](fe) Support iceberg REST catalog labels as the $labels table
(#69135)
### What problem does this PR solve?
Issue Number: #68916
Related PR: #69031
Problem Summary: The Iceberg REST catalog spec now lets a catalog return
`labels` with each table load (apache/iceberg#15750): catalog-owned,
per-request metadata such as ownership, classification or cost
attribution, at the table level (`object-labels`) and per field (keyed
by `field-id`). Labels are not part of the table metadata, have no
snapshots and are never written to `metadata.json`. Iceberg 1.12.0
exposes them on the loaded table via `SupportsLabels#labels()`. Doris
had no way to show them.
Add a `t$labels` system table for iceberg tables in a REST catalog:
```sql
SELECT * FROM rest_ctl.db.tbl$labels;
scope | field_id | field_name | key | value
object | NULL | NULL | owner | team-a
field | 2 | email | classification | pii
```
One row per label. `field_name` is resolved against the current table
schema and is NULL for a field id that is no longer in it (a dropped
column). The schema mirrors iceberg's proposed `LabelsTable`.
Implementation:
- Keep labels when Doris rebuilds a loaded table. iceberg's REST client
attaches them to the loaded `BaseTable` as a transient field; they are
not table metadata, so a `BaseTable` rebuilt from its `TableOperations`
must pass them explicitly.
`IcebergTableCache.TableLease#snapshotReadTable` (cached reads),
`IcebergStatementScope#snapshotReadTable` (statement reads) and
`IcebergScanPlanProvider#wrapTableForScan` (Kerberos FileIO wrap) used
constructors that reset labels to `Labels.EMPTY`, so every read saw no
labels; they now pass `baseTable.labels()`.
- `labels` is not a `MetadataTableType`, so it is handled next to the
SDK metadata tables rather than through `MetadataTableUtils`.
`IcebergConnectorMetadata` lists and resolves it only when the catalog
flavor is `rest` (no other catalog returns labels), and serves its fixed
schema and column handles without loading the base table.
- `IcebergScanPlanProvider` plans it as one split. A new split-package
helper `org.apache.iceberg.IcebergLabelsDataTask` builds an iceberg
`StaticDataTask` (package-private in the SDK) holding the label rows,
projected to the requested columns in BE's required-field order; its
metadata `DataFile` is synthesized, so no storage request is made. The
task is serialized like every other static metadata table, so BE's
`IcebergSysTableJniScanner` reads it with no BE change.
- The ordered system-table projection is extracted into
`projectSystemTableSchema` and shared with the existing path, and the
base-table load is split out of `resolveSysTable` as
`resolveSysBaseTable`.
- Labels are not versioned: like `$snapshots`/`$history`, the snapshot
pin of the handle is ignored, and `$labels` is exempt from the
resolved-empty-snapshot fence, so a table with no commit still shows its
labels.
Labels reflect the table load the statement used, so they follow the
iceberg table cache (`meta.cache.iceberg.table.ttl-second`); REFRESH
TABLE picks up new labels. Like the other system tables, a REST catalog
that requires server-side scan planning rejects `$labels`.
Co-authored-by: Isaac <[email protected]>
Co-authored-by: morningman <[email protected]>
---
fe/check/checkstyle/suppressions.xml | 2 +
.../iceberg/IcebergConnectorMetadata.java | 29 ++-
.../connector/iceberg/IcebergLabelsSysTable.java | 49 +++++
.../connector/iceberg/IcebergScanPlanProvider.java | 101 ++++++---
.../connector/iceberg/IcebergStatementScope.java | 2 +-
.../doris/connector/iceberg/IcebergTableCache.java | 2 +-
.../org/apache/iceberg/IcebergLabelsDataTask.java | 71 +++++++
.../IcebergConnectorMetadataSysTableTest.java | 50 +++++
.../iceberg/IcebergLabelsDataTaskTest.java | 143 +++++++++++++
.../iceberg/IcebergStatementScopeTest.java | 20 ++
.../connector/iceberg/IcebergTableCacheTest.java | 33 +++
.../iceberg/test_iceberg_rest_labels.out | 97 +++++++++
.../iceberg/test_iceberg_rest_labels.groovy | 233 +++++++++++++++++++++
13 files changed, 801 insertions(+), 31 deletions(-)
diff --git a/fe/check/checkstyle/suppressions.xml
b/fe/check/checkstyle/suppressions.xml
index 906a511b4be..96d40705bfe 100644
--- a/fe/check/checkstyle/suppressions.xml
+++ b/fe/check/checkstyle/suppressions.xml
@@ -79,6 +79,8 @@ under the License.
<suppress files="org[\\/]apache[\\/]iceberg[\\/]DeleteFileIndex\.java"
checks="[a-zA-Z0-9]*"/>
<!-- Iceberg package access is required to preserve historical schema
binding in DataTableScan. -->
<suppress
files="org[\\/]apache[\\/]iceberg[\\/]SchemaAwareDataTableScan\.java"
checks="[a-zA-Z0-9]*"/>
+ <!-- Iceberg package access is required to build a StaticDataTask for REST
catalog labels; other checks apply. -->
+ <suppress
files="org[\\/]apache[\\/]iceberg[\\/]IcebergLabelsDataTask\.java"
checks="ImportControl"/>
<!-- ignore gensrc/thrift/ExternalTableSchema.thrift -->
<suppress files=".*thrift/schema/external/.*" checks=".*"/>
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index b8b415f6f88..b571662d2b0 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -533,6 +533,12 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
// parse ITS schema (e.g. t$snapshots ->
committed_at/snapshot_id/...). Mirrors legacy
// IcebergSysExternalTable.getSysIcebergTable +
getOrCreateSchemaCacheValue; the enable.mapping.*
// flags are threaded by the shared buildTableSchema ->
parseSchema (deviation 5).
+ if (IcebergLabelsSysTable.isLabels(iceHandle)) {
+ // $labels is not an iceberg metadata table: its schema is
fixed and needs no table load (the
+ // base table's existence was already checked by
getTableHandle).
+ return new ConnectorTableSchema(iceHandle.getTableName(),
+ parseSchema(IcebergLabelsSysTable.SCHEMA), "ICEBERG",
Collections.emptyMap());
+ }
Table sysTable = loadSysTable(session, iceHandle);
return buildTableSchema(iceHandle.getTableName(), sysTable,
sysTable.schema(), false);
}
@@ -833,6 +839,9 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
// Mirror getTableSchema: wrap the remote load in the auth context. A
sys handle resolves the
// metadata-table columns (t$snapshots -> committed_at/...) so the
generic scan node can look up
// its pruned sys-table slots by name; a data handle resolves the base
table's columns.
+ if (IcebergLabelsSysTable.isLabels(iceHandle)) {
+ return buildColumnHandles(IcebergLabelsSysTable.SCHEMA, false);
+ }
Table table = iceHandle.isSystemTable() ? loadSysTable(session,
iceHandle) : loadTable(session, iceHandle);
return buildColumnHandles(table.schema(), !iceHandle.isSystemTable());
}
@@ -1919,6 +1928,9 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
* excluding it here would be a capability regression vs master. Unlike
every other entry, its scan takes
* BE's native reader rather than the JNI serialized-split path (see
* {@code IcebergScanPlanProvider.doPlanPositionDeletesSystemTableScan}).
+ *
+ * <p>A REST catalog additionally exposes {@code labels} ({@link
IcebergLabelsSysTable}), which is not a
+ * {@code MetadataTableType}: its rows are the catalog labels returned
with the table load.
*/
@Override
public List<String> listSupportedSysTables(ConnectorSession session,
@@ -1927,6 +1939,9 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
for (MetadataTableType type : MetadataTableType.values()) {
names.add(type.name().toLowerCase(Locale.ROOT));
}
+ if (supportsLabels()) {
+ names.add(IcebergLabelsSysTable.NAME);
+ }
return Collections.unmodifiableList(names);
}
@@ -1969,13 +1984,16 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
/**
* Whether iceberg exposes a system table named {@code sysName}
(case-insensitive). Mirrors
* {@code IcebergSysTable.SUPPORTED_SYS_TABLES}: every {@code
MetadataTableType}, {@code POSITION_DELETES}
- * included (see {@link #listSupportedSysTables}). A {@code null} name is
simply not exposed (returns
- * false, not NPE).
+ * included (see {@link #listSupportedSysTables}), plus {@code labels} for
a REST catalog. A {@code null}
+ * name is simply not exposed (returns false, not NPE).
*/
- private static boolean isSupportedSysTable(String sysName) {
+ private boolean isSupportedSysTable(String sysName) {
if (sysName == null) {
return false;
}
+ if (IcebergLabelsSysTable.NAME.equalsIgnoreCase(sysName)) {
+ return supportsLabels();
+ }
for (MetadataTableType type : MetadataTableType.values()) {
if (type.name().equalsIgnoreCase(sysName)) {
return true;
@@ -1984,6 +2002,11 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
return false;
}
+ /** Only a REST catalog returns catalog labels with the table load, so
only it exposes {@code $labels}. */
+ private boolean supportsLabels() {
+ return
IcebergCatalogProperties.TYPE_REST.equals(catalogProps.getFlavor());
+ }
+
// ========== Write / Transaction (P6.3) ==========
/**
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLabelsSysTable.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLabelsSysTable.java
new file mode 100644
index 00000000000..a088e4d515e
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLabelsSysTable.java
@@ -0,0 +1,49 @@
+// 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.doris.connector.iceberg;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+
+/**
+ * The {@code t$labels} system table: the catalog-provided labels of an
iceberg table loaded from a REST catalog
+ * (apache/iceberg#15750), one row per object-level or field-level label.
+ *
+ * <p>Unlike every other iceberg system table it is not a {@link
org.apache.iceberg.MetadataTableType}: labels
+ * come from the REST load response, not from the table metadata, are not
versioned with snapshots, and only a
+ * REST catalog returns them. Rows reflect the table load the statement read,
so they follow the iceberg table
+ * cache like the rest of the table's state. The schema mirrors iceberg's
proposed {@code LabelsTable}.
+ */
+final class IcebergLabelsSysTable {
+ static final String NAME = "labels";
+
+ static final Schema SCHEMA = new Schema(
+ Types.NestedField.required(1, "scope", Types.StringType.get()),
+ Types.NestedField.optional(2, "field_id", Types.IntegerType.get()),
+ Types.NestedField.optional(3, "field_name",
Types.StringType.get()),
+ Types.NestedField.required(4, "key", Types.StringType.get()),
+ Types.NestedField.required(5, "value", Types.StringType.get()));
+
+ private IcebergLabelsSysTable() {
+ }
+
+ /** Whether {@code handle} is a {@code $labels} handle; sys handles carry
the lower-cased name. */
+ static boolean isLabels(IcebergTableHandle handle) {
+ return NAME.equals(handle.getSysTableName());
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index fe1f1fd1da0..acbfe7f4587 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -47,12 +47,14 @@ import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BatchScan;
import org.apache.iceberg.ContentScanTask;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataTask;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.DeleteFileIndex;
import org.apache.iceberg.FileContent;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.IcebergLabelsDataTask;
import org.apache.iceberg.ManifestContent;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
@@ -977,6 +979,9 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private List<ConnectorScanRange> doPlanSystemTableScan(IcebergTableHandle
handle,
List<ConnectorColumnHandle> columns, Optional<ConnectorExpression>
filter,
ConnectorSession session) {
+ if (IcebergLabelsSysTable.isLabels(handle)) {
+ return doPlanLabelsSystemTableScan(handle, columns, session);
+ }
Table metadataTable = resolveSysTable(session, handle);
if (isPositionDeletesSysTable(handle)) {
return doPlanPositionDeletesSystemTableScan(handle, metadataTable,
columns, filter, session);
@@ -998,27 +1003,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// hand from the requested columns.
List<String> projectedColumns = requestedColumnNames(columns);
if (!projectedColumns.isEmpty()) {
- Schema metadataSchema = metadataTable.schema();
- List<NestedField> projectedFields = new
ArrayList<>(projectedColumns.size());
- Set<Integer> seenFieldIds = new HashSet<>();
- for (String columnName : projectedColumns) {
- NestedField field = metadataSchema.findField(columnName);
- if (field == null) {
- throw new RuntimeException("Column " + columnName
- + " not found in iceberg system table schema " +
metadataTable.name());
- }
- if (seenFieldIds.add(field.fieldId())) {
- projectedFields.add(field);
- }
- }
- // Iceberg returns split_offsets only when the last offset is
smaller than file_size_in_bytes.
- // Without that dependency, even valid offsets such as [4, 2048]
are hidden as null. Append it
- // after the requested columns so BE's positional reads keep their
original order.
- if (seenFieldIds.contains(DataFile.SPLIT_OFFSETS.fieldId())
- && seenFieldIds.add(DataFile.FILE_SIZE.fieldId())) {
-
projectedFields.add(metadataSchema.findField(DataFile.FILE_SIZE.fieldId()));
- }
- scan = scan.project(new Schema(projectedFields));
+ scan = scan.project(projectSystemTableSchema(
+ metadataTable.schema(), projectedColumns,
metadataTable.name()));
}
List<ConnectorScanRange> ranges = new ArrayList<>();
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
@@ -1043,6 +1029,58 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
&&
MetadataTableType.POSITION_DELETES.name().equalsIgnoreCase(handle.getSysTableName());
}
+ /**
+ * The system-table schema projected to the requested columns, in the
order BE lists them in
+ * required_fields: BE's {@code IcebergSysTableJniScanner} reads projected
rows positionally. Built by hand
+ * because {@code scan.select(names)} would re-emit the fields in
table-schema order. Any dependent metadata
+ * fields are appended after the requested columns.
+ */
+ private static Schema projectSystemTableSchema(Schema schema, List<String>
projectedColumns,
+ String tableName) {
+ List<NestedField> projectedFields = new
ArrayList<>(projectedColumns.size());
+ Set<Integer> seenFieldIds = new HashSet<>();
+ for (String columnName : projectedColumns) {
+ NestedField field = schema.findField(columnName);
+ if (field == null) {
+ throw new RuntimeException("Column " + columnName
+ + " not found in iceberg system table schema " +
tableName);
+ }
+ if (seenFieldIds.add(field.fieldId())) {
+ projectedFields.add(field);
+ }
+ }
+ // Iceberg returns split_offsets only when the last offset is smaller
than file_size_in_bytes.
+ // Without that dependency, even valid offsets such as [4, 2048] are
hidden as null. Append it
+ // after the requested columns so BE's positional reads keep their
original order.
+ if (seenFieldIds.contains(DataFile.SPLIT_OFFSETS.fieldId())
+ && seenFieldIds.add(DataFile.FILE_SIZE.fieldId())) {
+
projectedFields.add(schema.findField(DataFile.FILE_SIZE.fieldId()));
+ }
+ return new Schema(projectedFields);
+ }
+
+ /**
+ * Plan {@code t$labels}: one serialized split whose {@code
StaticDataTask} carries every catalog label of the
+ * base table, read by BE's JNI scanner like any other static metadata
table. Labels are not versioned, so
+ * the handle's snapshot pin is ignored (as for {@code $snapshots}/{@code
$history}); predicates are left to
+ * the scan node's conjuncts, the row count being bounded by the catalog
response.
+ */
+ private List<ConnectorScanRange>
doPlanLabelsSystemTableScan(IcebergTableHandle handle,
+ List<ConnectorColumnHandle> columns, ConnectorSession session) {
+ Table base = resolveSysBaseTable(session, handle);
+ List<String> projectedColumns = requestedColumnNames(columns);
+ Schema projectedSchema = projectedColumns.isEmpty()
+ ? IcebergLabelsSysTable.SCHEMA
+ : projectSystemTableSchema(IcebergLabelsSysTable.SCHEMA,
projectedColumns,
+ base.name() + "$" + IcebergLabelsSysTable.NAME);
+ DataTask task = IcebergLabelsDataTask.of(
+ base, IcebergLabelsSysTable.SCHEMA, projectedSchema,
SYS_TABLE_DUMMY_PATH);
+ return Collections.singletonList(new IcebergScanRange.Builder()
+ .path(SYS_TABLE_DUMMY_PATH)
+
.serializedSplit(IcebergSystemTableSerialization.serializeToBase64(task))
+ .build());
+ }
+
/**
* Plan a {@code $position_deletes} scan. Port of legacy
* {@code IcebergScanNode.doGetPositionDeletesSystemTableSplits} (upstream
#65135).
@@ -1328,6 +1366,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (!handle.isSystemTable()) {
return false;
}
+ if (IcebergLabelsSysTable.isLabels(handle)) {
+ // Catalog labels come with the table load, not from a snapshot,
so a table with no commit has them.
+ return true;
+ }
MetadataTableType type =
MetadataTableType.from(handle.getSysTableName());
// Only metadata_log_entries has a creation row before the first data
snapshot; every other metadata
// table must preserve the resolved-empty fence or a concurrent first
append becomes visible.
@@ -3398,9 +3440,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
BaseTable baseTable = (BaseTable) table;
TableOperations rawOps = baseTable.operations();
- // Keep the table's metrics reporter: every scan of a Kerberos catalog
runs on this wrapper.
+ // Keep the table's metrics reporter and REST catalog labels: every
scan of a Kerberos catalog runs on this
+ // wrapper.
return new BaseTable(new IcebergAuthenticatedTableOperations(
- rawOps, new IcebergAuthenticatedFileIO(rawOps.io(), auth)),
table.name(), baseTable.reporter());
+ rawOps, new IcebergAuthenticatedFileIO(rawOps.io(), auth)),
table.name(), baseTable.reporter(),
+ baseTable.labels());
}
/**
@@ -3413,6 +3457,13 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* task serialization) in ONE scope — no nested wrap here.
*/
private Table resolveSysTable(ConnectorSession session, IcebergTableHandle
handle) {
+ return MetadataTableUtils.createMetadataTableInstance(
+ resolveSysBaseTable(session, handle),
+ MetadataTableType.from(handle.getSysTableName()));
+ }
+
+ /** The base table a system-table scan reads, shared with metadata
binding; see {@link #resolveSysTable}. */
+ private Table resolveSysBaseTable(ConnectorSession session,
IcebergTableHandle handle) {
IcebergCatalogOps ops = catalogOpsResolver.apply(session);
// Keep the raw base shared with metadata binding and ordinary scan
properties. The caller already owns
// the auth scope, and avoiding a fresh load prevents system-table
slots and rows crossing generations.
@@ -3428,9 +3479,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
() -> ops.loadTable(handle.getDbName(),
handle.getTableName())),
() -> ops.loadTable(handle.getDbName(),
handle.getTableName()));
rejectServerSideScanPlanning(base, handle);
- return MetadataTableUtils.createMetadataTableInstance(
- base,
- MetadataTableType.from(handle.getSysTableName()));
+ return base;
}
/** This catalog's engine-owned storage services (see {@link
ConnectorContext#getStorageContext()}). */
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergStatementScope.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergStatementScope.java
index 6bb44ce3612..9d687e68b82 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergStatementScope.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergStatementScope.java
@@ -292,7 +292,7 @@ final class IcebergStatementScope {
TableMetadata metadata = operations.current();
// Keep IO, encryption and location behavior from the raw table while
pinning only metadata.
return new BaseTable(new IcebergSnapshotTableOperations(operations,
metadata),
- table.name(), baseTable.reporter());
+ table.name(), baseTable.reporter(), baseTable.labels());
}
/**
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
index b727259d491..ddc89637b8e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java
@@ -314,7 +314,7 @@ final class IcebergTableCache {
? operations.current()
: TableMetadataParser.fromJson(snapshotMetadataLocation,
snapshotMetadataJson);
return new BaseTable(new
IcebergSnapshotTableOperations(operations, metadata),
- table.name(), baseTable.reporter());
+ table.name(), baseTable.reporter(), baseTable.labels());
}
private TableLease tryBorrow() {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/IcebergLabelsDataTask.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/IcebergLabelsDataTask.java
new file mode 100644
index 00000000000..e8ae091c1c6
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/IcebergLabelsDataTask.java
@@ -0,0 +1,71 @@
+// 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.iceberg;
+
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Builds the single in-memory {@link DataTask} that carries a table's REST
catalog labels as rows.
+ *
+ * <p>Labels (apache/iceberg#15750) are returned by the catalog with each
table load and are not part of the
+ * table metadata, so no {@link MetadataTableType} can produce them. This
split-package helper builds an iceberg
+ * {@link StaticDataTask} directly (its constructors are package-private) so
the task serializes and is read by
+ * BE's {@code IcebergSysTableJniScanner} exactly like the SDK's own static
metadata tables. The metadata file is
+ * synthesized rather than built from an {@code InputFile}, which would issue
a storage HEAD for its length.
+ */
+public final class IcebergLabelsDataTask {
+ public static final String OBJECT_SCOPE = "object";
+ public static final String FIELD_SCOPE = "field";
+
+ private IcebergLabelsDataTask() {
+ }
+
+ /**
+ * Returns a task with one row per label: {@code (scope, field_id,
field_name, key, value)}. Field labels
+ * resolve {@code field_name} against the current table schema, so a label
on a dropped field keeps its
+ * {@code field_id} with a null {@code field_name}.
+ */
+ public static DataTask of(Table table, Schema labelsSchema, Schema
projectedSchema, String path) {
+ Preconditions.checkState(table instanceof SupportsLabels,
+ "Iceberg table %s does not expose catalog labels",
table.name());
+ Labels labels = ((SupportsLabels) table).labels();
+ Schema tableSchema = table.schema();
+ List<StructLike> rows = new ArrayList<>();
+ for (Map.Entry<String, String> label :
labels.objectLabels().entrySet()) {
+ rows.add(StaticDataTask.Row.of(OBJECT_SCOPE, null, null,
label.getKey(), label.getValue()));
+ }
+ for (FieldLabel field : labels.fields()) {
+ String fieldName = tableSchema.findColumnName(field.fieldId());
+ for (Map.Entry<String, String> label : field.labels().entrySet()) {
+ rows.add(StaticDataTask.Row.of(
+ FIELD_SCOPE, field.fieldId(), fieldName,
label.getKey(), label.getValue()));
+ }
+ }
+ DataFile metadataFile =
DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(path)
+ .withFileSizeInBytes(0L)
+ .withRecordCount(rows.size())
+ .withFormat(FileFormat.METADATA)
+ .build();
+ return new StaticDataTask(metadataFile, labelsSchema, projectedSchema,
rows.toArray(new StructLike[0]));
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataSysTableTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataSysTableTest.java
index 74d4b178756..d354d7eecc0 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataSysTableTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataSysTableTest.java
@@ -80,6 +80,12 @@ public class IcebergConnectorMetadataSysTableTest {
return new IcebergConnectorMetadata(ops,
IcebergCatalogProperties.of(Collections.emptyMap()), ctx);
}
+ private static IcebergConnectorMetadata
restMetadataWith(RecordingIcebergCatalogOps ops) {
+ Map<String, String> props = new HashMap<>();
+ props.put(IcebergCatalogProperties.ICEBERG_CATALOG_TYPE,
IcebergCatalogProperties.TYPE_REST);
+ return new IcebergConnectorMetadata(ops,
IcebergCatalogProperties.of(props), new RecordingConnectorContext());
+ }
+
private static IcebergTableHandle baseHandle() {
return new IcebergTableHandle("db1", "t1");
}
@@ -577,4 +583,48 @@ public class IcebergConnectorMetadataSysTableTest {
.getSysTableHandle(null, baseHandle(), "").isPresent(),
"an empty sys name must yield Optional.empty()");
}
+
+ // ---------------------------------------------------------------------
+ // $labels (REST catalog labels, not a MetadataTableType)
+ // ---------------------------------------------------------------------
+
+ @Test
+ public void listSupportedSysTablesAddsLabelsForRestCatalogOnly() {
+ // WHY: only a REST catalog returns labels with the table load;
elsewhere t$labels would always be
+ // empty, so it is not offered. MUTATION: listing labels
unconditionally -> non-REST list grows -> red.
+ List<String> expectedRest = expectedSupported();
+ expectedRest.add("labels");
+ Assertions.assertEquals(expectedRest,
+ restMetadataWith(new
RecordingIcebergCatalogOps()).listSupportedSysTables(null, baseHandle()));
+ Assertions.assertEquals(expectedSupported(),
+ metadataWith(new
RecordingIcebergCatalogOps()).listSupportedSysTables(null, baseHandle()));
+ }
+
+ @Test
+ public void getSysTableHandleResolvesLabelsForRestCatalogOnly() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ Optional<ConnectorTableHandle> rest =
restMetadataWith(ops).getSysTableHandle(null, baseHandle(), "LABELS");
+
+ Assertions.assertTrue(rest.isPresent());
+ Assertions.assertEquals("labels", ((IcebergTableHandle)
rest.get()).getSysTableName());
+ Assertions.assertTrue(ops.log.isEmpty(), "resolving t$labels must not
touch the catalog seam");
+ Assertions.assertFalse(metadataWith(new RecordingIcebergCatalogOps())
+ .getSysTableHandle(null, baseHandle(), "labels").isPresent());
+ }
+
+ @Test
+ public void getTableSchemaForLabelsHandleIsFixedAndLoadsNothing() {
+ // WHY: labels has no iceberg metadata table, so the generic sys path
(MetadataTableType.from -> null)
+ // cannot build its schema; the schema is fixed and needs no base
load. MUTATION: falling through to
+ // loadSysTable -> NPE / a catalog round-trip -> red.
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ IcebergConnectorMetadata md = restMetadataWith(ops);
+ IcebergTableHandle sysHandle = (IcebergTableHandle)
md.getSysTableHandle(null, baseHandle(), "labels").get();
+
+ Assertions.assertEquals(Arrays.asList("scope", "field_id",
"field_name", "key", "value"),
+ columnNames(md.getTableSchema(null, sysHandle)));
+ Assertions.assertEquals(Arrays.asList("scope", "field_id",
"field_name", "key", "value"),
+ new ArrayList<>(md.getColumnHandles(null,
sysHandle).keySet()));
+ Assertions.assertTrue(ops.log.isEmpty(), "t$labels schema must not
load the base table");
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLabelsDataTaskTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLabelsDataTaskTest.java
new file mode 100644
index 00000000000..b15771c2d68
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergLabelsDataTaskTest.java
@@ -0,0 +1,143 @@
+// 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.doris.connector.iceberg;
+
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.DataTask;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.IcebergLabelsDataTask;
+import org.apache.iceberg.ImmutableFieldLabel;
+import org.apache.iceberg.ImmutableLabels;
+import org.apache.iceberg.Labels;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Tests for the {@code t$labels} rows ({@link IcebergLabelsDataTask}): one
row per object-level and field-level
+ * label, projected in the requested order, and carried intact through the
system-table serialization that BE's
+ * JNI scanner reads.
+ */
+public class IcebergLabelsDataTaskTest {
+
+ private static final Schema BASE_SCHEMA = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "email", Types.StringType.get()));
+
+ private static final Labels LABELS = ImmutableLabels.builder()
+ .objectLabels(Collections.singletonMap("owner", "team-a"))
+ .addFields(ImmutableFieldLabel.builder()
+ .fieldId(2)
+ .labels(Collections.singletonMap("classification", "pii"))
+ .build())
+ // Field 7 is not in the current schema (e.g. a dropped column):
keep its id, no name.
+ .addFields(ImmutableFieldLabel.builder()
+ .fieldId(7)
+ .labels(Collections.singletonMap("deprecated", "true"))
+ .build())
+ .build();
+
+ private static Table labeledTable(Labels labels) {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", Collections.emptyMap());
+ catalog.createNamespace(Namespace.of("db1"));
+ BaseTable table = (BaseTable) catalog.createTable(
+ TableIdentifier.of("db1", "t1"), BASE_SCHEMA,
PartitionSpec.unpartitioned());
+ return new BaseTable(table.operations(), table.name(),
table.reporter(), labels);
+ }
+
+ /** Reads every row eagerly: StaticDataTask re-wraps one projection per
row, so rows cannot be kept. */
+ private static List<List<Object>> read(DataTask task, int width) throws
IOException {
+ List<List<Object>> rows = new ArrayList<>();
+ try (CloseableIterable<StructLike> iterable = task.rows()) {
+ for (StructLike row : iterable) {
+ List<Object> values = new ArrayList<>(width);
+ for (int i = 0; i < width; i++) {
+ values.add(row.get(i, Object.class));
+ }
+ rows.add(values);
+ }
+ }
+ return rows;
+ }
+
+ @Test
+ public void emitsObjectAndFieldLabelRows() throws IOException {
+ DataTask task = IcebergLabelsDataTask.of(labeledTable(LABELS),
IcebergLabelsSysTable.SCHEMA,
+ IcebergLabelsSysTable.SCHEMA, "/dummyPath");
+
+ Assertions.assertEquals(Arrays.asList(
+ Arrays.asList("object", null, null, "owner", "team-a"),
+ Arrays.asList("field", 2, "email", "classification", "pii"),
+ Arrays.asList("field", 7, null, "deprecated", "true")),
+ read(task, 5));
+ Assertions.assertEquals(3L, task.file().recordCount());
+ }
+
+ @Test
+ public void projectsInRequestedOrder() throws IOException {
+ // BE reads the projected row positionally in required_fields order,
so the task must project to exactly
+ // the requested schema. MUTATION: building the task over the full
schema -> (scope, field_id) -> red.
+ Schema projected = new Schema(
+ IcebergLabelsSysTable.SCHEMA.findField("value"),
+ IcebergLabelsSysTable.SCHEMA.findField("scope"));
+ DataTask task = IcebergLabelsDataTask.of(labeledTable(LABELS),
IcebergLabelsSysTable.SCHEMA,
+ projected, "/dummyPath");
+
+ Assertions.assertEquals(Arrays.asList(
+ Arrays.asList("team-a", "object"),
+ Arrays.asList("pii", "field"),
+ Arrays.asList("true", "field")),
+ read(task, 2));
+ }
+
+ @Test
+ public void tableWithoutLabelsHasNoRows() throws IOException {
+ DataTask task = IcebergLabelsDataTask.of(labeledTable(Labels.EMPTY),
IcebergLabelsSysTable.SCHEMA,
+ IcebergLabelsSysTable.SCHEMA, "/dummyPath");
+
+ Assertions.assertTrue(read(task, 5).isEmpty());
+ }
+
+ @Test
+ public void rowsSurviveSystemTableSerialization() throws IOException {
+ // The task is shipped to BE as a serialized FileScanTask; labels must
be materialized in it, since the
+ // BE side has no catalog to ask. MUTATION: keeping a lazy reference
to the table's labels -> red.
+ DataTask task = IcebergLabelsDataTask.of(labeledTable(LABELS),
IcebergLabelsSysTable.SCHEMA,
+ IcebergLabelsSysTable.SCHEMA, "/dummyPath");
+
+ FileScanTask decoded =
IcebergSystemTableSerialization.deserializeFromBase64(
+ IcebergSystemTableSerialization.serializeToBase64(task));
+
+ Assertions.assertEquals(read(task, 5), read(decoded.asDataTask(), 5));
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergStatementScopeTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergStatementScopeTest.java
index 2df7d37c00b..dd35b63c8ae 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergStatementScopeTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergStatementScopeTest.java
@@ -21,6 +21,9 @@ import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
import org.apache.doris.thrift.TIcebergDeleteFileDesc;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.ImmutableLabels;
+import org.apache.iceberg.Labels;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
@@ -79,6 +82,23 @@ public class IcebergStatementScopeTest {
Assertions.assertEquals(1, loads.get(), "loaded once per statement");
}
+ @Test
+ public void sharedTableKeepsCatalogLabels() {
+ // The statement-scoped read view is a rebuilt BaseTable; REST catalog
labels must survive the rebuild
+ // because they are not part of the table metadata. MUTATION: dropping
labels on rebuild -> EMPTY -> red.
+ BaseTable raw = (BaseTable) mutableTable("t");
+ Labels labels = ImmutableLabels.builder()
+ .objectLabels(Collections.singletonMap("owner", "team-a"))
+ .build();
+ Table labeled = new BaseTable(raw.operations(), raw.name(),
raw.reporter(), labels);
+
+ Table shared = IcebergStatementScope.sharedTable(
+ new ScopeSession(7L, "q1", new TestStatementScope()), "db1",
"t", () -> labeled);
+
+ Assertions.assertNotSame(labeled, shared);
+ Assertions.assertEquals(labels, ((BaseTable) shared).labels());
+ }
+
@Test
public void sameStatementSharesOneWritableTable() {
ScopeSession session = new ScopeSession(7L, "q1", new
TestStatementScope());
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
index 8ee2b2e6db6..b3e82db5fe2 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergTableCacheTest.java
@@ -24,6 +24,9 @@ import org.apache.doris.connector.cache.JvmSizeUtils;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.ImmutableFieldLabel;
+import org.apache.iceberg.ImmutableLabels;
+import org.apache.iceberg.Labels;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -42,6 +45,7 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.metrics.LoggingMetricsReporter;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.JsonUtil;
import org.junit.jupiter.api.Assertions;
@@ -498,6 +502,35 @@ public class IcebergTableCacheTest {
return new BaseTable(new StaticTableOperations(refsOnly),
"lazy-snapshots");
}
+ @Test
+ public void snapshotReadTableKeepsCatalogLabels() {
+ // REST catalog labels live only on the loaded BaseTable (not in
metadata.json), so the statement view
+ // must carry them over. MUTATION: rebuilding with the 3-arg BaseTable
constructor -> Labels.EMPTY -> red.
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ TableMetadata metadata = TableMetadata.newTableMetadata(
+ schema, PartitionSpec.unpartitioned(), "file:///tmp/labeled",
Collections.emptyMap());
+ Labels labels = ImmutableLabels.builder()
+ .objectLabels(Collections.singletonMap("owner", "team-a"))
+ .addFields(ImmutableFieldLabel.builder()
+ .fieldId(1)
+ .labels(Collections.singletonMap("classification",
"pii"))
+ .build())
+ .build();
+ Table table = new BaseTable(new StaticTableOperations(metadata),
"labeled",
+ LoggingMetricsReporter.instance(), labels);
+
+ try (CatalogMetaCache owner = CatalogMetaCache.unmanaged()) {
+ IcebergTableCache cache = new IcebergTableCache(
+ owner, CacheSpec.ofWeight(true, 100L, 1000L, 10L * 1024L *
1024L),
+ ignored -> () -> { }, new IcebergCatalogResourceTracker());
+ try (IcebergTableCache.TableLease lease = cache.borrow(id(), () ->
table)) {
+ BaseTable statement = (BaseTable) lease.snapshotReadTable();
+ Assertions.assertNotSame(table, statement);
+ Assertions.assertEquals(labels, statement.labels());
+ }
+ }
+ }
+
@Test
public void ttlZeroDisablesCachingAlwaysLive() {
AtomicInteger loads = new AtomicInteger();
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_rest_labels.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_rest_labels.out
new file mode 100644
index 00000000000..060f94202fb
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_rest_labels.out
@@ -0,0 +1,97 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !desc --
+field_id int Yes true \N NONE
+field_name text Yes true \N NONE
+key text Yes true \N NONE
+scope text Yes true \N NONE
+value text Yes true \N NONE
+
+-- !all --
+field 2 email classification pii
+field 3 phone classification pii
+field 3 phone masking partial
+field 5 address.city classification location
+object \N \N cost.center cc-42
+object \N \N owner team-a
+object \N \N steward 数据平台组
+
+-- !projection --
+cc-42 cost.center \N object \N
+location classification address.city field 5
+partial masking phone field 3
+pii classification email field 2
+pii classification phone field 3
+team-a owner \N object \N
+数据平台组 steward \N object \N
+
+-- !one_column --
+classification
+classification
+classification
+cost.center
+masking
+owner
+steward
+
+-- !filter_field --
+email classification pii
+phone classification pii
+
+-- !filter_object --
+cost.center cc-42
+owner team-a
+steward 数据平台组
+
+-- !count --
+7
+
+-- !count_table --
+2
+
+-- !no_snapshot --
+object \N \N owner team-b
+
+-- !unlabeled --
+0
+
+-- !direct --
+0
+
+-- !vended --
+field 2 email classification pii
+field 3 phone classification pii
+field 3 phone masking partial
+field 5 address.city classification location
+object \N \N cost.center cc-42
+object \N \N owner team-a
+object \N \N steward 数据平台组
+
+-- !schema_evolution --
+field 2 email_address classification pii
+field 3 \N classification pii
+field 3 \N masking partial
+field 5 address.city classification location
+object \N \N cost.center cc-42
+object \N \N owner team-a
+object \N \N steward 数据平台组
+
+-- !time_travel --
+field 2 email_address classification pii
+field 3 \N classification pii
+field 3 \N masking partial
+field 5 address.city classification location
+object \N \N cost.center cc-42
+object \N \N owner team-a
+object \N \N steward 数据平台组
+
+-- !cached --
+owner team-b
+
+-- !uncached --
+owner team-c
+tier gold
+
+-- !refreshed --
+owner team-c
+tier gold
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_labels.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_labels.groovy
new file mode 100644
index 00000000000..3b6b8965e06
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_rest_labels.groovy
@@ -0,0 +1,233 @@
+// 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.
+
+import com.sun.net.httpserver.HttpExchange
+import com.sun.net.httpserver.HttpHandler
+import com.sun.net.httpserver.HttpServer
+import groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+
+import java.net.InetSocketAddress
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executors
+
+// t$labels shows the labels a REST catalog returns with each table load
(apache/iceberg#15750). The docker REST
+// fixture returns none (a fixture option is apache/iceberg#18047), so this
suite puts a small proxy in front of
+// it: the proxy forwards every request unchanged and adds this suite's labels
to the load-table responses, which
+// is all a labelling catalog does differently. Everything after the HTTP
response is the real path: the iceberg
+// REST client parses the labels, FE plans the labels split and BE reads it
through the JNI system-table scanner.
+suite("test_iceberg_rest_labels", "p0,external") {
+ if
(!"true".equalsIgnoreCase(context.config.otherConfigs.get("enableIcebergTest")))
{
+ return
+ }
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String restUri = "http://${externalEnvIp}:${restPort}";
+
+ // Labels the proxy adds, keyed by "<namespace>.<table>". A table without
an entry keeps the fixture's
+ // response as is, without labels.
+ Map<String, Map> labelsByTable = new ConcurrentHashMap<>()
+ // Lengths and hop-by-hop headers belong to each connection.
Accept-Encoding is dropped so that the fixture
+ // answers in plain JSON the proxy can rewrite.
+ Set<String> droppedRequestHeaders = ["host", "connection", "keep-alive",
"content-length", "transfer-encoding",
+ "expect", "accept-encoding"] as Set
+ def loadTablePath = ~'^/v1/namespaces/([^/]+)/tables/([^/]+)$'
+ def proxyExecutor = Executors.newFixedThreadPool(4)
+ HttpServer proxy = HttpServer.create(new InetSocketAddress("0.0.0.0", 0),
0)
+ proxy.setExecutor(proxyExecutor)
+ proxy.createContext("/", { HttpExchange exchange ->
+ try {
+ String rawPath = exchange.requestURI.rawPath
+ String rawQuery = exchange.requestURI.rawQuery
+ def upstream = (HttpURLConnection) new URL(restUri + rawPath +
(rawQuery == null ? "" : "?" + rawQuery))
+ .openConnection()
+ upstream.setRequestMethod(exchange.requestMethod)
+ upstream.setConnectTimeout(10000)
+ upstream.setReadTimeout(60000)
+ exchange.requestHeaders.each { String name, List<String> values ->
+ if (!droppedRequestHeaders.contains(name.toLowerCase())) {
+ values.each { upstream.addRequestProperty(name, it) }
+ }
+ }
+ byte[] requestBody = exchange.requestBody.bytes
+ if (requestBody.length > 0) {
+ upstream.setDoOutput(true)
+ upstream.outputStream.withCloseable { it.write(requestBody) }
+ }
+ int status = upstream.responseCode
+ InputStream upstreamBody = status >= 400 ? upstream.errorStream :
upstream.inputStream
+ byte[] body = upstreamBody == null ? new byte[0] :
upstreamBody.withCloseable { it.bytes }
+ def loadTable = loadTablePath.matcher(rawPath)
+ if ("GET" == exchange.requestMethod && status == 200 &&
loadTable.matches()) {
+ Map labels = labelsByTable.get(loadTable.group(1) + "." +
loadTable.group(2))
+ if (labels != null) {
+ def response = new JsonSlurper().parseText(new
String(body, "UTF-8"))
+ response.labels = labels
+ body = JsonOutput.toJson(response).getBytes("UTF-8")
+ }
+ }
+ // Pass back only the content type. The fixture's validators
describe the body before the rewrite: a
+ // forwarded ETag would let the iceberg client revalidate its
cached load and keep stale labels.
+ String contentType = upstream.getHeaderField("Content-Type")
+ if (contentType != null) {
+ exchange.responseHeaders.set("Content-Type", contentType)
+ }
+ exchange.sendResponseHeaders(status, body.length == 0 ? -1 :
body.length)
+ if (body.length > 0) {
+ exchange.responseBody.withCloseable { it.write(body) }
+ }
+ } finally {
+ exchange.close()
+ }
+ } as HttpHandler)
+ proxy.start()
+ try {
+ // FE loads the tables through the proxy, so it must be reachable from
FE; like the HTTP TVF suite, use
+ // httpTvfHost when the regression runner is not on the docker host.
+ String proxyHost = context.config.otherConfigs.get("httpTvfHost") ?:
externalEnvIp
+ String proxyUri = "http://${proxyHost}:${proxy.address.port}";
+
+ sql "drop catalog if exists test_iceberg_rest_labels"
+ sql """create catalog test_iceberg_rest_labels properties (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'${proxyUri}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.region' = 'us-east-1'
+ )"""
+ // With a zero table cache TTL every statement loads the table, and so
its labels, again.
+ sql "drop catalog if exists test_iceberg_rest_labels_nocache"
+ sql """create catalog test_iceberg_rest_labels_nocache properties (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'${proxyUri}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.region' = 'us-east-1',
+ 'meta.cache.iceberg.table.ttl-second' = '0'
+ )"""
+ // Vended credentials turn the cross-query table cache off altogether,
so this catalog reads the table that
+ // each statement loads instead of a cached one.
+ sql "drop catalog if exists test_iceberg_rest_labels_vended"
+ sql """create catalog test_iceberg_rest_labels_vended properties (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'${proxyUri}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.region' = 'us-east-1',
+ 'iceberg.rest.vended-credentials-enabled' = 'true'
+ )"""
+ // The same tables read straight from the fixture, which returns no
labels. The tables are created through
+ // it, so no proxied catalog caches a load before its labels are set.
+ sql "drop catalog if exists test_iceberg_rest_labels_direct"
+ sql """create catalog test_iceberg_rest_labels_direct properties (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'rest', 'uri' =
'${restUri}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.region' = 'us-east-1'
+ )"""
+
+ sql "create database if not exists
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db"
+ sql "drop table if exists
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled"
+ sql """create table
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled (
+ id int, email string, phone string, address struct<city: string,
zip: string>
+ )"""
+ sql """insert into
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled values
+ (1, '[email protected]', '555-0101', named_struct('city', 'Paris',
'zip', '75001')),
+ (2, '[email protected]', '555-0102', named_struct('city', 'Lyon',
'zip', '69001'))"""
+ sql "drop table if exists
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled_empty"
+ sql "create table
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled_empty (id
int, note string)"
+ sql "drop table if exists
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.unlabeled"
+ sql "create table
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.unlabeled (id int)"
+ sql "insert into
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.unlabeled values
(1), (2)"
+
+ // Field labels are keyed by field id, which the catalog assigns: read
the ids from the fixture.
+ def loaded = new JsonSlurper().parseText(
+ new
URL("${restUri}/v1/namespaces/test_iceberg_rest_labels_db/tables/labeled").getText(
+ connectTimeout: 10000, readTimeout: 10000))
+ def schema = loaded.metadata.schemas.find { it["schema-id"] ==
loaded.metadata["current-schema-id"] }
+ def emailId = schema.fields.find { it.name == "email" }.id
+ def phoneId = schema.fields.find { it.name == "phone" }.id
+ def cityId = schema.fields.find { it.name == "address"
}.type.fields.find { it.name == "city" }.id
+ labelsByTable.put("test_iceberg_rest_labels_db.labeled", [
+ "object-labels": ["owner": "team-a", "cost.center": "cc-42",
"steward": "数据平台组"],
+ "fields": [
+ ["field-id": emailId, "labels": ["classification":
"pii"]],
+ ["field-id": phoneId, "labels": ["classification":
"pii", "masking": "partial"]],
+ ["field-id": cityId, "labels": ["classification":
"location"]]]])
+ labelsByTable.put("test_iceberg_rest_labels_db.labeled_empty",
["object-labels": ["owner": "team-b"]])
+
+ sql "switch test_iceberg_rest_labels"
+ sql "use test_iceberg_rest_labels_db"
+ order_qt_desc "desc labeled\$labels"
+ // One row per label. A field label names its field from the current
schema, a nested one by its full path.
+ order_qt_all "select * from labeled\$labels"
+ // BE reads the split's rows by position in the order of the columns
it asks for, so another order or a
+ // subset of the columns must still put every value in its own column.
KEY is a reserved word in Doris,
+ // so the key column is quoted.
+ order_qt_projection "select value, `key`, field_name, scope, field_id
from labeled\$labels"
+ order_qt_one_column "select `key` from labeled\$labels"
+ order_qt_filter_field "select field_name, `key`, value from
labeled\$labels where scope = 'field' and value = 'pii'"
+ order_qt_filter_object "select `key`, value from labeled\$labels where
field_id is null"
+ // COUNT counts labels rather than taking the table's record count:
two records, seven labels.
+ order_qt_count "select count(*) from labeled\$labels"
+ order_qt_count_table "select count(*) from labeled"
+ // Labels come with the table load, not with a snapshot: a table
without any commit has them too.
+ order_qt_no_snapshot "select * from labeled_empty\$labels"
+ // A table the catalog returns no labels for, and the labelled table
read without the proxy.
+ order_qt_unlabeled "select count(*) from unlabeled\$labels"
+ order_qt_direct "select count(*) from
test_iceberg_rest_labels_direct.test_iceberg_rest_labels_db.labeled\$labels"
+ // Without the table cache the labels come straight from the
statement's own table load.
+ order_qt_vended "select * from
test_iceberg_rest_labels_vended.test_iceberg_rest_labels_db.labeled\$labels"
+
+ // field_name follows the field id through schema evolution: a renamed
column keeps its labels under its new
+ // name, and a dropped one keeps its field_id with a NULL field_name.
+ sql "alter table labeled rename column email email_address"
+ sql "alter table labeled drop column phone"
+ order_qt_schema_evolution "select * from labeled\$labels"
+ // Labels are not versioned: a time-travel pin on t$labels is ignored,
so the rows are those of the current
+ // load and field names still come from the current schema, not from
the schema of the pinned snapshot.
+ def snapshotId = sql("select snapshot_id from
labeled\$snapshots")[0][0]
+ order_qt_time_travel "select * from labeled\$labels for version as of
${snapshotId}"
+
+ // Labels belong to the table load, so they follow the table cache
like the rest of the table: the cached
+ // catalog keeps the labels it loaded until REFRESH TABLE, the
uncached one sees new labels at once.
+ labelsByTable.put("test_iceberg_rest_labels_db.labeled_empty",
["object-labels": ["owner": "team-c", "tier": "gold"]])
+ order_qt_cached "select `key`, value from labeled_empty\$labels"
+ order_qt_uncached "select `key`, value from
test_iceberg_rest_labels_nocache.test_iceberg_rest_labels_db.labeled_empty\$labels"
+ sql "refresh table labeled_empty"
+ order_qt_refreshed "select `key`, value from labeled_empty\$labels"
+
+ // Only a REST catalog returns labels with the table load, so no other
catalog offers t$labels.
+ sql "drop catalog if exists test_iceberg_rest_labels_hadoop"
+ sql """create catalog test_iceberg_rest_labels_hadoop properties (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'hadoop',
+ 'warehouse' = 's3a://warehouse/test_iceberg_rest_labels_hadoop',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
's3.region' = 'us-east-1'
+ )"""
+ sql "create database if not exists
test_iceberg_rest_labels_hadoop.test_iceberg_rest_labels_db"
+ sql "drop table if exists
test_iceberg_rest_labels_hadoop.test_iceberg_rest_labels_db.hadoop_table"
+ sql "create table
test_iceberg_rest_labels_hadoop.test_iceberg_rest_labels_db.hadoop_table (id
int)"
+ test {
+ sql "select * from
test_iceberg_rest_labels_hadoop.test_iceberg_rest_labels_db.hadoop_table\$labels"
+ exception "Unknown sys table 'hadoop_table\$labels'"
+ }
+
+ // The proxy stops with the suite, so its catalogs go too; the tables
stay readable through the direct one.
+ sql "switch internal"
+ sql "drop catalog test_iceberg_rest_labels"
+ sql "drop catalog test_iceberg_rest_labels_nocache"
+ sql "drop catalog test_iceberg_rest_labels_vended"
+ } finally {
+ proxy.stop(0)
+ proxyExecutor.shutdownNow()
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
