924060929 commented on code in PR #64688:
URL: https://github.com/apache/doris/pull/64688#discussion_r3548422746
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java:
##########
@@ -112,32 +333,1357 @@ public Optional<ConnectorTableHandle> getTableHandle(
public ConnectorTableSchema getTableSchema(
ConnectorSession session, ConnectorTableHandle handle) {
IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
- String dbName = iceHandle.getDbName();
- String tableName = iceHandle.getTableName();
+ if (iceHandle.isSystemTable()) {
+ // System (metadata) table: load the base table and build the
iceberg metadata-table, then
+ // 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).
+ Table sysTable = loadSysTable(iceHandle);
+ return buildTableSchema(iceHandle.getTableName(), sysTable,
sysTable.schema());
+ }
+ // Mirror legacy IcebergMetadataOps.loadTable: wrap the remote load in
the auth context. The schema
+ // + table-property assembly is pure (operates on the already-loaded
Table).
+ Table table = loadTable(iceHandle);
+ return buildTableSchema(iceHandle.getTableName(), table,
table.schema());
+ }
+
+ /**
+ * Returns the schema AS OF {@code snapshot.getSchemaId()} (the pinned
schema version, for time-travel reads
+ * under schema evolution), or the LATEST schema when there is no pinned
schema id (null snapshot or
+ * {@code schemaId < 0}). Mirrors legacy {@code IcebergUtils.getSchema}:
{@code table.schemas().get(schemaId)}
+ * when the id is set and a current snapshot exists, else {@code
table.schema()}. Shares
+ * {@link #buildTableSchema} with the latest path so the two cannot drift.
+ */
+ @Override
+ public ConnectorTableSchema getTableSchema(
+ ConnectorSession session, ConnectorTableHandle handle,
ConnectorMvccSnapshot snapshot) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ if (iceHandle.isSystemTable()) {
+ // A metadata table has a FIXED schema, independent of
snapshot/schema-version (t$snapshots
+ // always exposes committed_at/snapshot_id/...; legacy has no
schema-at-snapshot for sys
+ // tables). The time-travel pin (deviation 1) selects which rows
the SCAN reads (T05), not the
+ // schema, so delegate to the latest path, which builds the
metadata-table schema.
+ return getTableSchema(session, handle);
+ }
+ if (snapshot == null || snapshot.getSchemaId() < 0) {
+ return getTableSchema(session, handle);
+ }
+ Table table = loadTable(iceHandle);
+ Schema schema;
+ if (table.currentSnapshot() == null) {
+ // Empty table: legacy getSchema falls back to the latest schema
(NEWEST_SCHEMA_ID path).
+ schema = table.schema();
+ } else {
+ schema = table.schemas().get((int) snapshot.getSchemaId());
+ if (schema == null) {
+ // Defensive: a pinned id absent from table.schemas() (legacy
would NPE) -> latest.
+ schema = table.schema();
+ }
+ }
+ return buildTableSchema(iceHandle.getTableName(), table, schema);
+ }
+
+ /**
+ * Assembles the {@link ConnectorTableSchema} for {@code table} from
{@code schema} (the latest schema, or a
+ * historical schema for a time-travel read). The {@code
iceberg.format-version} / {@code location} /
+ * {@code iceberg.partition-spec} properties are table-level (not
schema-versioned). Factored out so the
+ * latest and at-snapshot paths share ONE assembly.
+ */
+ private ConnectorTableSchema buildTableSchema(String tableName, Table
table, Schema schema) {
+ List<ConnectorColumn> columns = parseSchema(schema);
- Table table = catalog.loadTable(TableIdentifier.of(dbName, tableName));
- Schema icebergSchema = table.schema();
- List<ConnectorColumn> columns = parseSchema(icebergSchema);
+ // Append the iceberg v3 row-lineage hidden columns (_row_id /
_last_updated_sequence_number) for
+ // format-version >= 3 tables, mirroring legacy
IcebergUtils.appendRowLineageColumnsForV3 — invoked
+ // unconditionally (format-gated) from
IcebergExternalTable.getFullSchema. They are BIGINT, nullable,
+ // non-key, hidden, and carry a reserved Doris field id (matched
BE-side); convertColumn re-applies
+ // setIsVisible(false)/setUniqueId. Appended AFTER the data columns
(legacy append order). Metadata
+ // (system) tables report format-version 2
(BaseMetadataTable.properties() is empty), so the gate
+ // naturally excludes them — matching legacy, which injects lineage
only for data tables.
+ if (getFormatVersion(table) >= ICEBERG_ROW_LINEAGE_MIN_VERSION) {
+ columns.add(new ConnectorColumn(ICEBERG_ROW_ID_COL,
ConnectorType.of("BIGINT"),
+ "", true, null,
false).invisible().withUniqueId(ICEBERG_ROW_ID_FIELD_ID));
+ columns.add(new
ConnectorColumn(ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_COL,
ConnectorType.of("BIGINT"),
+ "", true, null,
false).invisible().withUniqueId(ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER_FIELD_ID));
+ }
Map<String, String> tableProps = new HashMap<>();
tableProps.putAll(table.properties());
- tableProps.put("iceberg.format-version",
- String.valueOf(table.spec().specId() >= 0 ? 2 : 1));
+ // SHOW CREATE TABLE render hints under neutral reserved keys (fe-core
strips them from the
+ // rendered PROPERTIES and emits them as LOCATION / PARTITION BY /
ORDER BY). They replace the
+ // previously-injected location / iceberg.format-version /
iceberg.partition-spec keys: those were
+ // never read by fe-core and would leak into the rendered
PROPERTIES(...) (legacy iceberg SHOW
+ // CREATE dumped only the raw table.properties()). format-version
stays available via the
+ // getFormatVersion(table) gate above for the row-lineage columns; it
is not a user property.
if (table.location() != null) {
- tableProps.put("location", table.location());
+ tableProps.put(ConnectorTableSchema.SHOW_LOCATION_KEY,
table.location());
+ }
+ String partitionClause = buildShowPartitionClause(table);
+ if (!partitionClause.isEmpty()) {
+ tableProps.put(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY,
partitionClause);
+ }
+ String sortClause = buildShowSortClause(table);
+ if (!sortClause.isEmpty()) {
+ tableProps.put(ConnectorTableSchema.SHOW_SORT_CLAUSE_KEY,
sortClause);
}
if (!table.spec().isUnpartitioned()) {
- tableProps.put("iceberg.partition-spec", table.spec().toString());
+ // Generic FE partition-column contract: post-cutover,
PluginDrivenExternalTable derives the
+ // table's partition columns SOLELY from a "partition_columns" CSV
property (toSchemaCacheValue),
+ // the same key MaxCompute/paimon emit. Mirror legacy
IcebergUtils.loadTableSchemaCacheValue:
+ // walk the CURRENT spec, resolve each partition field's SOURCE
column name (NO identity filter,
+ // NO dedupe), lowercased to match parseSchema's lowercased column
names (fromRemoteColumnName is
+ // identity for iceberg, so the FE consumer looks the names up
case-sensitively).
+ List<String> partitionColumns = new ArrayList<>();
+ for (PartitionField field : table.spec().fields()) {
+ Types.NestedField source =
table.schema().findField(field.sourceId());
+ if (source != null) {
+
partitionColumns.add(source.name().toLowerCase(Locale.ROOT));
+ }
+ }
+ if (!partitionColumns.isEmpty()) {
+ tableProps.put("partition_columns", String.join(",",
partitionColumns));
+ }
}
return new ConnectorTableSchema(tableName, columns, "ICEBERG",
tableProps);
}
+ /**
+ * Pre-renders the Doris {@code PARTITION BY LIST (...) ()} clause from
the iceberg {@link PartitionSpec}
+ * for SHOW CREATE TABLE (the FE plugin-driven path has no live iceberg
API). Mirrors legacy
+ * {@code IcebergExternalTable.getPartitionSpecSql}: void -> skipped,
identity -> bare column,
+ * {@code bucket[N]}/{@code truncate[W]}/{@code year}/{@code month}/{@code
day}/{@code hour} -> the
+ * matching Doris partition function. Returns "" for an unpartitioned
table or no renderable field.
+ */
+ private String buildShowPartitionClause(Table table) {
+ PartitionSpec spec = table.spec();
+ if (spec == null || spec.isUnpartitioned()) {
+ return "";
+ }
+ List<String> fields = new ArrayList<>();
+ for (PartitionField field : spec.fields()) {
+ String colName = table.schema().findColumnName(field.sourceId());
+ if (colName == null) {
+ continue;
+ }
+ org.apache.iceberg.transforms.Transform<?, ?> t =
field.transform();
+ if (t.isVoid()) {
+ continue;
+ }
+ String quotedCol = "`" + colName + "`";
+ if (t.isIdentity()) {
+ fields.add(quotedCol);
+ } else {
+ String transformStr = t.toString();
+ if (transformStr.startsWith("bucket[")) {
+ int n = Integer.parseInt(transformStr.substring(7,
transformStr.length() - 1));
+ fields.add("BUCKET(" + n + ", " + quotedCol + ")");
+ } else if (transformStr.startsWith("truncate[")) {
+ int w = Integer.parseInt(transformStr.substring(9,
transformStr.length() - 1));
+ fields.add("TRUNCATE(" + w + ", " + quotedCol + ")");
+ } else if ("year".equals(transformStr)) {
+ fields.add("YEAR(" + quotedCol + ")");
+ } else if ("month".equals(transformStr)) {
+ fields.add("MONTH(" + quotedCol + ")");
+ } else if ("day".equals(transformStr)) {
+ fields.add("DAY(" + quotedCol + ")");
+ } else if ("hour".equals(transformStr)) {
+ fields.add("HOUR(" + quotedCol + ")");
+ } else {
+ LOG.warn("Unsupported Iceberg partition transform '{}' on
column '{}', "
+ + "skipped in SHOW CREATE TABLE.", transformStr,
colName);
+ }
+ }
+ }
+ if (fields.isEmpty()) {
+ return "";
+ }
+ return "PARTITION BY LIST (" + String.join(", ", fields) + ") ()";
+ }
+
+ /**
+ * Pre-renders the Doris {@code ORDER BY (...)} clause from the iceberg
{@link SortOrder} for SHOW
+ * CREATE TABLE. Mirrors legacy {@code
IcebergExternalTable.getSortOrderSql} + {@code SortFieldInfo.toSql}
+ * ({@code `col` ASC|DESC NULLS FIRST|LAST}). Returns "" when the table is
unsorted.
+ */
+ static String buildShowSortClause(Table table) {
+ SortOrder sortOrder = table.sortOrder();
+ if (sortOrder == null || sortOrder.isUnsorted() ||
sortOrder.fields().isEmpty()) {
+ return "";
+ }
+ List<String> sortItems = new ArrayList<>();
+ for (org.apache.iceberg.SortField sortField : sortOrder.fields()) {
+ String columnName =
table.schema().findColumnName(sortField.sourceId());
+ if (columnName != null) {
+ boolean isAscending = sortField.direction() !=
org.apache.iceberg.SortDirection.DESC;
+ boolean isNullFirst = sortField.nullOrder() ==
org.apache.iceberg.NullOrder.NULLS_FIRST;
+ sortItems.add("`" + columnName + "`"
+ + (isAscending ? " ASC" : " DESC")
+ + " NULLS " + (isNullFirst ? "FIRST" : "LAST"));
+ }
+ }
+ return "ORDER BY (" + String.join(", ", sortItems) + ")";
+ }
+
+ /** Loads the iceberg {@link Table} through the seam, wrapped in the
FE-injected auth context (Kerberos UGI). */
+ private Table loadTable(IcebergTableHandle handle) {
+ try {
+ return context.executeAuthenticated(
+ () -> catalogOps.loadTable(handle.getDbName(),
handle.getTableName()));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to load table, error message
is:" + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Loads the iceberg metadata (system) table for {@code handle} through
the seam, wrapped in the
+ * FE-injected auth context (Kerberos UGI). Mirrors legacy
+ * {@code IcebergSysExternalTable.getSysIcebergTable}: load the base table
by its BASE coordinates,
+ * then build the metadata table via {@code
MetadataTableUtils.createMetadataTableInstance}. Both the
+ * base load and the (in-memory) metadata-table build run inside ONE
{@code executeAuthenticated} so
+ * the auth scope covers the remote base load. {@code
handle.getSysTableName()} is the lower-cased name
+ * already validated by {@code getSysTableHandle}, so {@code
MetadataTableType.from} (case-insensitive)
+ * never returns null.
+ */
+ private Table loadSysTable(IcebergTableHandle handle) {
+ try {
+ return context.executeAuthenticated(() -> {
+ Table base = catalogOps.loadTable(handle.getDbName(),
handle.getTableName());
+ return MetadataTableUtils.createMetadataTableInstance(
+ base,
MetadataTableType.from(handle.getSysTableName()));
+ });
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to load table, error message
is:" + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Column handles keyed by (lowercased) column name, mirroring {@code
PaimonConnectorMetadata}. The generic
+ * {@code PluginDrivenScanNode.buildColumnHandles} looks each query slot
up here by name, so the provider
+ * receives the PRUNED set of requested columns — which the T06 field-id
schema dictionary keys its
+ * {@code current_schema_id = -1} entry off (the CI #969249 fix: the
dict's top-level names == the BE
+ * scan-slot names BY CONSTRUCTION). The field id is the iceberg {@code
NestedField.fieldId()} (a permanent
+ * invariant). The name is lowercased with {@code Locale.ROOT} to
byte-match {@link #parseSchema} (so the
+ * handle key == the Doris slot name).
+ */
+ @Override
+ public Map<String, ConnectorColumnHandle> getColumnHandles(
+ ConnectorSession session, ConnectorTableHandle handle) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ // 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.
+ Table table = iceHandle.isSystemTable() ? loadSysTable(iceHandle) :
loadTable(iceHandle);
+ List<Types.NestedField> fields = table.schema().columns();
+ Map<String, ConnectorColumnHandle> handles = new
LinkedHashMap<>(fields.size());
+ for (Types.NestedField field : fields) {
+ String name = field.name().toLowerCase(Locale.ROOT);
+ handles.put(name, new IcebergColumnHandle(name, field.fieldId()));
+ }
+ return handles;
+ }
+
+ /**
+ * Table-level row count, surfaced to the FE optimizer via {@code
PluginDrivenExternalTable.fetchRowCount}
+ * (without this override the connector inherits {@code
ConnectorStatisticsOps}'s {@code Optional.empty()},
+ * so every iceberg table reports rowCount -1 -> CBO collapses cardinality
to 1 and disables join reorder).
+ * Mirrors {@code PaimonConnectorMetadata.getTableStatistics} in
STRUCTURE, but uses the legacy iceberg
+ * FORMULA ({@code IcebergUtils.getIcebergRowCount}: currentSnapshot
summary {@code total-records -
+ * total-position-deletes}). Parity decisions:
+ * <ul>
+ * <li>System tables -> empty: legacy {@code
IcebergSysExternalTable.fetchRowCount} is unconditionally
+ * UNKNOWN; a sys handle would otherwise load the BASE table and
misreport its data row count for a
+ * metadata table. (This is a deliberate divergence from paimon,
which reports sys-table counts.)</li>
+ * <li>{@code rowCount > 0} gate: legacy data-table consumer is {@code
rowCount > 0 ? rowCount : UNKNOWN},
+ * but the NEW consumer takes the value whenever {@code >= 0}, so a
0-row table would wrongly report 0.
+ * Collapsing {@code <= 0} to empty pins the "0 -> UNKNOWN"
semantics here (matches paimon).</li>
+ * <li>Any failure degrades to empty (best effort): a statistics miss
must never break query planning.</li>
+ * </ul>
+ */
+ @Override
+ public Optional<ConnectorTableStatistics> getTableStatistics(
+ ConnectorSession session, ConnectorTableHandle handle) {
+ IcebergTableHandle iceHandle = (IcebergTableHandle) handle;
+ if (iceHandle.isSystemTable()) {
+ return Optional.empty();
+ }
+ long rowCount;
+ try {
+ rowCount = computeRowCount(loadTable(iceHandle));
+ } catch (Exception e) {
+ LOG.warn("Failed to compute Iceberg row count for {}.{}",
+ iceHandle.getDbName(), iceHandle.getTableName(), e);
+ return Optional.empty();
+ }
+ if (rowCount > 0) {
+ return Optional.of(new ConnectorTableStatistics(rowCount, -1));
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Row count from the current snapshot summary: {@code total-records -
total-position-deletes} (legacy
+ * {@code IcebergUtils.getIcebergRowCount}). NOT the COUNT(*)-pushdown
formula in
+ * {@code IcebergScanPlanProvider.getCountFromSnapshot} — that one gates
on equality deletes and honors the
+ * dangling-delete session var (scan cardinality), which would
over-degrade table statistics. Empty table
+ * (no current snapshot) -> -1, which the caller maps to UNKNOWN.
+ */
+ private static long computeRowCount(Table table) {
+ Snapshot snapshot = table.currentSnapshot();
+ if (snapshot == null) {
+ return -1;
+ }
+ Map<String, String> summary = snapshot.summary();
+ // Null-guard (upstream 32a2651f66b, #64648): compaction / replace /
overwrite snapshots may omit a
+ // total-* counter, and the pre-fix Long.parseLong(null) NPE-d. -1 ->
caller maps to UNKNOWN. NOTE:
+ // intentionally NOT the pushdown getCountFromSummary — this
table-stats path deliberately omits the
+ // equality-delete gate (see the method javadoc), so only the NPE is
fixed here, not the semantics.
+ String totalRecords = summary.get(TOTAL_RECORDS);
+ String positionDeletes = summary.get(TOTAL_POSITION_DELETES);
+ if (totalRecords == null || positionDeletes == null) {
+ return -1;
+ }
+ return Long.parseLong(totalRecords) - Long.parseLong(positionDeletes);
Review Comment:
**Table-statistics row count drops the equality-delete gate** (medium,
CONFIRMED). `computeRowCount` returns `totalRecords - positionDeletes`
unconditionally, but legacy `getIcebergRowCount → getCountFromSummary(summary,
true)` returns UNKNOWN when `total-equality-deletes != "0"` (master
IcebergUtils.java:234, called from the stats path :1206). For a merge-on-read /
CDC-upsert table with equality deletes, this over-counts (the equality-deleted
rows are not subtracted), so `SHOW TABLE STATS` and CBO cardinality are
inflated — order-of-magnitude for update-heavy tables — skewing join order.
Note the asymmetry: the sibling COUNT(*) pushdown path
(IcebergScanPlanProvider.getCountFromSummary) keeps the full gate; only the
stats path regressed. The null-guard from #64648 (32a2651f66b, an ancestor of
this branch) was kept but the equality gate it sits next to was not.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogFactory.java:
##########
@@ -0,0 +1,689 @@
+// 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.doris.connector.api.DorisConnectorException;
+import org.apache.doris.connector.cache.CacheSpec;
+import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties;
+import org.apache.doris.filesystem.properties.StorageProperties;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.aws.AssumeRoleAwsClientFactory;
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.AwsProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
+import org.apache.iceberg.rest.auth.OAuth2Properties;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Pure, testable assembly core for the Iceberg connector flavor switch — the
iceberg-SDK-specific bits
+ * that stay in the connector. Mirrors the role of {@code
PaimonCatalogFactory}: a stateless static
+ * holder whose methods are PURE (they read only the supplied props — no env,
no clock, no live
+ * catalog), which is what makes them unit-testable offline.
+ *
+ * <p>P6.1 (this task) holds only the flavor resolution: {@link
#resolveFlavor(Map)} (the lower-cased
+ * {@code iceberg.catalog.type}) and {@link #resolveCatalogImpl(String)} (the
catalog-impl class name
+ * for the five {@code CatalogUtil}-built flavors plus the two bespoke ones).
The full per-flavor
+ * property / Hadoop-{@code Configuration} / {@code HiveConf} assembly
currently dropped by the
+ * skeleton — ported from the fe-core {@code AbstractIcebergProperties} + each
+ * {@code Iceberg*MetaStoreProperties#initCatalog} — lands in a later task
(P6-T05/T06/T07); this task
+ * is a structural inversion only, with no behavior change.
+ *
+ * <p>Note: {@code s3tables} and {@code dlf} are listed here for completeness,
but legacy does NOT build
+ * them via {@code CatalogUtil.buildIcebergCatalog} (s3tables hand-builds an
{@code S3TablesClient};
+ * dlf uses {@code new DLFCatalog().setConf(..).initialize(..)}). Routing them
through the impl-name
+ * path is the existing skeleton behavior, preserved verbatim here; the
bespoke instantiation fixes are
+ * P6-T06 / P6-T07.
+ */
+public final class IcebergCatalogFactory {
+
+ // Manifest-cache derivation defaults — mirror fe-core
IcebergExternalCatalog so the
+ // meta.cache.iceberg.manifest.* -> io.manifest.cache-enabled derivation
matches legacy exactly.
+ private static final boolean DEFAULT_MANIFEST_CACHE_ENABLE = false;
+ private static final long DEFAULT_MANIFEST_CACHE_TTL_SECOND = 48L * 60 *
60;
+ private static final long DEFAULT_MANIFEST_CACHE_CAPACITY = 1024L;
+
+ // Mirror of legacy AWSGlueMetaStoreBaseProperties.ENDPOINT_PATTERN:
extracts the region from a glue
+ // endpoint host (e.g. glue.us-east-1.amazonaws.com /
glue-fips.us-east-1.api.aws) when no explicit
+ // glue.region is set, before falling back to us-east-1.
+ private static final Pattern GLUE_ENDPOINT_PATTERN = Pattern.compile(
+
"^(?:https?://)?(?:glue|glue-fips)\\.([a-z0-9-]+)\\.(?:api\\.aws|amazonaws\\.com)$");
+
+ // Region-field aliases scanned to propagate client.region when NO
fe-filesystem S3 storage is bound
+ // (e.g. REST vended credentials: no static AK/SK/role, so
S3FileSystemProvider.supports is false and
+ // chosenS3 is empty). Mirrors the legacy
AbstractIcebergProperties.toFileIOProperties chosen==null
+ // fallback (getRegionFromProperties). The raw s3.region copied by
buildBaseCatalogProperties is inert
+ // because iceberg S3FileIO reads client.region, not s3.region.
+ private static final String[] S3_REGION_ALIASES = {"s3.region",
"aws.region", "region", "client.region"};
Review Comment:
**Region alias set narrowed vs legacy** (medium, CONFIRMED). This fallback
(used when no S3 storage binds — exactly the REST vended-credentials path, per
the comment at :78) scans only `{s3.region, aws.region, region,
client.region}`, dropping aliases legacy honored: `AWS_REGION` (common) and the
REST-specific `iceberg.rest.signing-region` / `rest.signing-region` (master
AbstractS3CompatibleProperties.getRegionFromProperties +
S3Properties.java:87-88 scan every isRegionField alias across
S3/OSS/COS/OBS/Minio). A REST catalog with vended credentials whose region was
set via a dropped alias now emits no `client.region` → S3FileIO falls through
to the AWS DefaultAwsRegionProviderChain and fails with 'Unable to load region'
on hosts without ambient region. The :78 comment's claim to 'mirror' legacy
getRegionFromProperties is inaccurate — either widen the alias list or reuse
the legacy scanner.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RowLevelDmlRegistry.java:
##########
@@ -0,0 +1,55 @@
+// 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.nereids.trees.plans.commands;
+
+import org.apache.doris.catalog.TableIf;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Registry of {@link RowLevelDmlTransform}s. The dispatching DML commands
consult this instead of testing the
+ * target table type, so the reverse {@code instanceof} dispatch is
consolidated here.
+ *
+ * <p>Explicit static registration (no {@code ServiceLoader}) — avoids the
thread-context-classloader pitfalls
+ * seen with SPI loaders. Today the single entry is {@link
IcebergRowLevelDmlTransform}, whose {@code handles}
+ * is {@code instanceof IcebergExternalTable}; at P6.6 cutover the predicate
can become a capability check.</p>
+ */
+public final class RowLevelDmlRegistry {
+
+ private static final List<RowLevelDmlTransform> TRANSFORMS =
+ ImmutableList.of(new IcebergRowLevelDmlTransform());
Review Comment:
**Generic gate, iceberg-shaped body** (design/altitude). The registry admits
any plugin table whose connector declares `WriteOperation.DELETE/MERGE`, but
the only registered transform synthesizes
`IcebergDeleteCommand`/`IcebergUpdateCommand`/`IcebergMergeCommand` that bind
`UnboundSlot(Column.ICEBERG_ROWID_COL)` (`__DORIS_ICEBERG_ROWID_COL__`) and
iceberg branch/projection algebra. This is the instanceof→capability inversion
the migration is meant to remove, just hidden behind a capability. The next
row-level-DML connector (hive ACID in P7, or paimon adding DELETE — both named
in the plan docs as the evolution trigger) that declares the capability gets
routed into the iceberg transform and produces a plan referencing a rowid
column it never exposes → unresolvable-slot error. Worth a follow-up to make
the transform connector-selected (or the row-id column SPI-declared) before
another connector trips it.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java:
##########
@@ -78,63 +531,533 @@ private Catalog getOrCreateCatalog() {
}
private Catalog createCatalog() {
- String catalogType =
properties.get(IcebergConnectorProperties.ICEBERG_CATALOG_TYPE);
- if (catalogType == null || catalogType.isEmpty()) {
+ String flavor = IcebergCatalogFactory.resolveFlavor(properties);
+ if (flavor == null) {
throw new DorisConnectorException(
"Missing '" +
IcebergConnectorProperties.ICEBERG_CATALOG_TYPE + "' property");
}
- Map<String, String> catalogProps = new HashMap<>(properties);
- String catalogImpl = resolveCatalogImpl(catalogType);
- catalogProps.put(CatalogProperties.CATALOG_IMPL, catalogImpl);
- // Iceberg SDK does not allow both "type" and "catalog-impl"
- catalogProps.remove(CatalogUtil.ICEBERG_CATALOG_TYPE);
+ Optional<S3CompatibleFileSystemProperties> chosenS3 =
+
IcebergCatalogFactory.chooseS3Compatible(context.getStorageProperties());
+ String catalogName =
IcebergCatalogFactory.resolveCatalogName(properties, flavor,
context.getCatalogName());
- Configuration conf = buildHadoopConf(catalogProps);
- String catalogName = context.getCatalogName();
+ // s3tables is bespoke: it is NOT built via
CatalogUtil.buildIcebergCatalog. Legacy
+ // IcebergS3TablesMetaStoreProperties hand-builds an S3TablesClient
and calls the 3-arg
+ // S3TablesCatalog.initialize(name, opts, client). Routed before the
CatalogUtil flavor switch.
+ if (IcebergConnectorProperties.TYPE_S3_TABLES.equals(flavor)) {
+ return createS3TablesCatalog(catalogName, chosenS3);
+ }
- LOG.info("Creating Iceberg catalog '{}' with type='{}', impl='{}'",
- catalogName, catalogType, catalogImpl);
+ // dlf is bespoke too: legacy IcebergAliyunDLFMetaStoreProperties
builds a hive-compatible DLFCatalog with
+ // a DataLakeConfig-keyed Configuration and an OSS-backed S3FileIO,
NOT via CatalogUtil.buildIcebergCatalog.
+ if (IcebergConnectorProperties.TYPE_DLF.equals(flavor)) {
+ return createDlfCatalog(catalogName, chosenS3);
+ }
- return CatalogUtil.buildIcebergCatalog(catalogName, catalogProps,
conf);
+ Map<String, String> catalogProps =
+ IcebergCatalogFactory.buildCatalogProperties(properties,
flavor, chosenS3);
+ Map<String, String> storageHadoopConfig = buildStorageHadoopConfig();
+
+ Configuration conf;
+ switch (flavor) {
+ case IcebergConnectorProperties.TYPE_HMS: {
+ // Reuse the shared metastore-spi parser (Q2=B bindForType):
iceberg passes its own flavor token
+ // so the metastore-spi never learns iceberg.catalog.type.
Only toHiveConfOverrides is used
+ // (iceberg HMS does NOT call paimon's validate(); it does not
require a warehouse). The external
+ // hive.conf.resources hive-site.xml is resolved FE-side and
seeded as the HiveConf base.
+ Map<String, String> hiveConfFiles =
context.loadHiveConfResources(
+ IcebergCatalogFactory.firstNonBlank(properties,
"hive.conf.resources"));
+ HmsMetaStoreProperties hms = (HmsMetaStoreProperties)
MetaStoreProviders.bindForType(
+ IcebergConnectorProperties.TYPE_HMS, properties,
storageHadoopConfig);
+ conf = IcebergCatalogFactory.assembleHiveConf(hiveConfFiles,
+ hms.toHiveConfOverrides(context.getEnvironment()
+
.getOrDefault("hive_metastore_client_timeout_second", "10")));
+ break;
+ }
+ case IcebergConnectorProperties.TYPE_GLUE:
+ // Legacy IcebergGlueMetaStoreProperties builds the catalog
with conf=null.
+ conf = null;
+ break;
+ case IcebergConnectorProperties.TYPE_JDBC:
+ maybeRegisterJdbcDriver();
+ conf =
IcebergCatalogFactory.buildHadoopConfiguration(properties, storageHadoopConfig);
+ break;
+ default:
+ // rest / hadoop: a storage Configuration from the
fe-filesystem-bound storage + raw
+ // fs./dfs./hadoop. passthrough.
+ conf =
IcebergCatalogFactory.buildHadoopConfiguration(properties, storageHadoopConfig);
+ break;
+ }
+
+ LOG.info("Creating Iceberg catalog '{}' flavor='{}' impl='{}'",
+ catalogName, flavor,
catalogProps.get(CatalogProperties.CATALOG_IMPL));
+ return buildCatalogAuthenticated(flavor,
+ () -> CatalogUtil.buildIcebergCatalog(catalogName,
catalogProps, conf));
}
/**
- * Resolve the Iceberg catalog implementation class from the catalog type
string.
+ * Creates the bespoke {@code s3tables} catalog, mirroring legacy {@code
IcebergS3TablesMetaStoreProperties}: a
+ * hand-built {@link S3TablesClient} (region + credentials + optional
{@code s3tables.endpoint} override + the
+ * s3tables-SDK http config) is passed to the 3-arg {@code
S3TablesCatalog.initialize(name, opts, client)} —
+ * NOT to {@code CatalogUtil.buildIcebergCatalog}. The 2-arg {@code
initialize(name, opts)} is intentionally
+ * avoided: its {@code DefaultS3TablesAwsClientFactory} honors only a
{@code client.credentials-provider} class
+ * and would silently drop static {@code s3.access-key-id}/{@code
s3.secret-access-key} (falling back to the
+ * SDK default chain). A bound S3-compatible storage is required to derive
region + credentials; a missing one
+ * (or a blank region) fails loud here, before any AWS call.
*/
- private static String resolveCatalogImpl(String catalogType) {
- switch (catalogType.toLowerCase()) {
- case "rest":
- return "org.apache.iceberg.rest.RESTCatalog";
- case "hms":
- return "org.apache.iceberg.hive.HiveCatalog";
- case "glue":
- return "org.apache.iceberg.aws.glue.GlueCatalog";
- case "hadoop":
- return "org.apache.iceberg.hadoop.HadoopCatalog";
- case "jdbc":
- return "org.apache.iceberg.jdbc.JdbcCatalog";
- case "s3tables":
- return "software.amazon.s3tables.iceberg.S3TablesCatalog";
- case "dlf":
- return "org.apache.doris.connector.iceberg.dlf.DLFCatalog";
- default:
- throw new DorisConnectorException(
- "Unknown iceberg.catalog.type: " + catalogType
- + ". Supported types: rest, hms, glue, hadoop,
jdbc, s3tables, dlf");
+ private Catalog createS3TablesCatalog(String catalogName,
Optional<S3CompatibleFileSystemProperties> chosenS3) {
+ if (!chosenS3.isPresent()) {
+ throw new DorisConnectorException(
+ "Iceberg s3tables catalog requires S3-compatible storage
properties (region + credentials)");
Review Comment:
**s3tables now hard-fails without explicit S3 credentials** (medium,
CONFIRMED). This throw fires when no S3 storage binds, and
`S3FileSystemProvider.supports()` requires an explicit access key / role ARN /
credentials-provider-type (fe-filesystem-s3 :65-73). So a legacy-valid s3tables
catalog using the AWS default credential chain — only `s3.region` + a
`warehouse` ARN on an EC2 instance-profile host — can no longer be created.
Legacy `IcebergS3TablesMetaStoreProperties` went through
`createAwsCredentialsProvider` whose PROVIDER_CHAIN branch returns the SDK
default chain (region required, creds not). Consider allowing the bind when a
region is present and letting the default chain resolve credentials at use time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
