This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 57e41e92c28 [fix](fe) Allow table access through Iceberg JDBC V0
catalogs (#69452)
57e41e92c28 is described below
commit 57e41e92c28d0699b0d777cb173da716f601f70d
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 23 06:47:22 2026 +0800
[fix](fe) Allow table access through Iceberg JDBC V0 catalogs (#69452)
### What problem does this PR solve?
Issue Number: N/A
Problem Summary:
Iceberg's `JdbcCatalog` implements `ViewCatalog` even when its JDBC
metadata schema is V0, which cannot store views. The SDK throws
`UnsupportedOperationException` for V0 `listViews` and `viewExists`.
Doris calls these methods when listing tables and resolving whether an
ordinary table is a view, so an existing V0 catalog can fail table
discovery and queries despite containing no views.
Treat unsupported JDBC view probes as an empty view list or `false`, and
reuse the same probe path when filtering table names. Preserve working
JDBC view support, explicit view-operation errors, failures from other
catalog types, and SQL/authorization errors. Reading V0 does not require
migrating the catalog to V1.
Add six unit tests using the actual SDK V0 guards and negative coverage.
Extend the JDBC regression suite with an isolated PostgreSQL V0 database
and a SELECT-only reader, covering namespace/table listing, schema
inspection, scans, snapshots, refresh, and time travel. The fixture
database is separate because the SDK inspects columns across PostgreSQL
schemas.
### Release note
Fix table discovery and access through Iceberg JDBC catalogs that use
the V0 metadata schema without view support.
### Check List (For Author)
- Test
- [x] Unit Test: Iceberg connector reactor build and tests: 1,493 tests,
zero failures/errors, one existing live REST connectivity test skipped.
FE Checkstyle passes. Before the fix, the new tests reproduce two
failures from the actual SDK V0 guards.
- [x] Regression test: `test_iceberg_jdbc_catalog`, including the new V0
reader and existing PostgreSQL/MySQL cases. Local runtime uses existing
FE/BE binaries with the modified connector class compiled from master,
PostgreSQL 16, MySQL 8, and MinIO.
- [x] Manual test: verify the V0 table has no `iceberg_type` column and
the reader has SELECT but lacks INSERT, UPDATE, and schema CREATE
privileges.
- Behavior changed:
- [x] Yes. Unsupported JDBC view probes no longer prevent ordinary table
operations.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/connector/iceberg/IcebergCatalogOps.java | 28 ++--
.../iceberg/IcebergJdbcViewProbeTest.java | 143 +++++++++++++++++++++
.../iceberg/test_iceberg_jdbc_catalog.groovy | 83 ++++++++++++
3 files changed, 245 insertions(+), 9 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
index 4968a141ab2..0e60d18ab41 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
@@ -45,6 +45,7 @@ import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Term;
import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.jdbc.JdbcCatalog;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.view.View;
@@ -60,6 +61,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -376,12 +378,7 @@ public interface IcebergCatalogOps {
.collect(Collectors.toList());
// iceberg's listTables also returns views, so subtract the view
names when the catalog is a
// (view-enabled) ViewCatalog — mirrors legacy
IcebergMetadataOps.listTableNames.
- if (!isViewCatalogEnabled()) {
- return tableNames;
- }
- List<String> views = resolveViewCatalog().listViews(ns).stream()
- .map(TableIdentifier::name)
- .collect(Collectors.toList());
+ List<String> views = listViewNames(dbName);
if (views.isEmpty()) {
return tableNames;
}
@@ -398,9 +395,9 @@ public interface IcebergCatalogOps {
if (!isViewCatalogEnabled()) {
return Collections.emptyList();
}
- return resolveViewCatalog().listViews(toNamespace(dbName)).stream()
+ return probeViews(() ->
resolveViewCatalog().listViews(toNamespace(dbName)).stream()
.map(TableIdentifier::name)
- .collect(Collectors.toList());
+ .collect(Collectors.toList()), Collections.emptyList());
}
@Override
@@ -415,7 +412,7 @@ public interface IcebergCatalogOps {
if (!isViewCatalogEnabled()) {
return false;
}
- return resolveViewCatalog().viewExists(toTableIdentifier(dbName,
viewName));
+ return probeViews(() ->
resolveViewCatalog().viewExists(toTableIdentifier(dbName, viewName)), false);
}
@Override
@@ -878,6 +875,19 @@ public interface IcebergCatalogOps {
return (SupportsNamespaces) catalog;
}
+ private <T> T probeViews(Supplier<T> probe, T noViews) {
+ try {
+ return probe.get();
+ } catch (UnsupportedOperationException e) {
+ // JdbcCatalog implements ViewCatalog even for V0, which
cannot store views. The SDK
+ // rejects these probes before SQL; do not require a V1
migration merely to read tables.
+ if (resolveViewCatalog() instanceof JdbcCatalog) {
+ return noViews;
+ }
+ throw e;
+ }
+ }
+
/** View filtering is on iff a view catalog is resolvable and (for
REST) views are enabled. */
private boolean isViewCatalogEnabled() {
if (resolveViewCatalog() == null) {
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.java
new file mode 100644
index 00000000000..854b09dff81
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergJdbcViewProbeTest.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.doris.connector.iceberg.IcebergCatalogOps.CatalogBackedIcebergCatalogOps;
+
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ForbiddenException;
+import org.apache.iceberg.jdbc.JdbcCatalog;
+import org.apache.iceberg.jdbc.UncheckedSQLException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+public class IcebergJdbcViewProbeTest {
+ private static final Namespace NAMESPACE = Namespace.of("db");
+ private static final TableIdentifier TABLE = TableIdentifier.of(NAMESPACE,
"table");
+ private static final TableIdentifier VIEW = TableIdentifier.of(NAMESPACE,
"view");
+
+ private IcebergCatalogOps ops(Catalog catalog) {
+ return new CatalogBackedIcebergCatalogOps(catalog, false, false, true,
Optional.empty());
+ }
+
+ @Test
+ public void v0ListsTablesWithoutViewSupport() {
+ Assertions.assertEquals(Collections.singletonList("table"), ops(new
V0JdbcCatalog()).listTableNames("db"));
+ }
+
+ @Test
+ public void v0HasNoViews() {
+ IcebergCatalogOps ops = ops(new V0JdbcCatalog());
+ Assertions.assertEquals(Collections.emptyList(),
ops.listViewNames("db"));
+ Assertions.assertFalse(ops.viewExists("db", "table"));
+ Assertions.assertFalse(ops.viewExists("db", "missing"));
+ }
+
+ @Test
+ public void explicitV0ViewOperationsStillFail() {
+ IcebergCatalogOps ops = ops(new V0JdbcCatalog());
+ Assertions.assertThrows(UnsupportedOperationException.class, () ->
ops.loadView("db", "view"));
+ Assertions.assertThrows(UnsupportedOperationException.class, () ->
ops.dropView("db", "view"));
+ }
+
+ @Test
+ public void jdbcViewSupportIsNotDisabledByCatalogType() {
+ IcebergCatalogOps ops = ops(new V0JdbcCatalog() {
+ @Override
+ public List<TableIdentifier> listTables(Namespace namespace) {
+ return Arrays.asList(TABLE, VIEW);
+ }
+
+ @Override
+ public List<TableIdentifier> listViews(Namespace namespace) {
+ return Collections.singletonList(VIEW);
+ }
+
+ @Override
+ public boolean viewExists(TableIdentifier identifier) {
+ return VIEW.equals(identifier);
+ }
+ });
+ Assertions.assertEquals(Collections.singletonList("table"),
ops.listTableNames("db"));
+ Assertions.assertEquals(Collections.singletonList("view"),
ops.listViewNames("db"));
+ Assertions.assertTrue(ops.viewExists("db", "view"));
+ Assertions.assertFalse(ops.viewExists("db", "table"));
+ }
+
+ @Test
+ public void jdbcProbeFailuresAreNotHidden() {
+ for (RuntimeException failure : Arrays.asList(
+ new ForbiddenException("Access denied"), new
UncheckedSQLException("Connection failed"))) {
+ IcebergCatalogOps ops = ops(new V0JdbcCatalog() {
+ @Override
+ public List<TableIdentifier> listViews(Namespace namespace) {
+ throw failure;
+ }
+
+ @Override
+ public boolean viewExists(TableIdentifier identifier) {
+ throw failure;
+ }
+ });
+ Assertions.assertSame(failure,
Assertions.assertThrows(failure.getClass(),
+ () -> ops.listTableNames("db")));
+ Assertions.assertSame(failure,
Assertions.assertThrows(failure.getClass(),
+ () -> ops.listViewNames("db")));
+ Assertions.assertSame(failure,
Assertions.assertThrows(failure.getClass(),
+ () -> ops.viewExists("db", "view")));
+ }
+ }
+
+ @Test
+ public void unsupportedViewsInOtherCatalogsAreNotHidden() {
+ UnsupportedOperationException failure = new
UnsupportedOperationException("Unsupported view operation");
+ FakeIcebergViewCatalog catalog = new FakeIcebergViewCatalog() {
+ @Override
+ public List<TableIdentifier> listViews(Namespace namespace) {
+ throw failure;
+ }
+
+ @Override
+ public boolean viewExists(TableIdentifier identifier) {
+ throw failure;
+ }
+ };
+ IcebergCatalogOps ops = ops(catalog);
+ Assertions.assertSame(failure,
Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> ops.listTableNames("db")));
+ Assertions.assertSame(failure,
Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> ops.listViewNames("db")));
+ Assertions.assertSame(failure,
Assertions.assertThrows(UnsupportedOperationException.class,
+ () -> ops.viewExists("db", "view")));
+ }
+
+ // Keep the SDK's actual V0 listViews/viewExists guards; only table
listing avoids a live JDBC connection.
+ private static class V0JdbcCatalog extends JdbcCatalog {
+ @Override
+ public List<TableIdentifier> listTables(Namespace namespace) {
+ return Collections.singletonList(TABLE);
+ }
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
index 10ab5bc8cb5..729b82428f0 100644
---
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy
@@ -290,7 +290,88 @@ suite("test_iceberg_jdbc_catalog", "p0,external") {
check("restored_namespace", "SELECT * FROM jdbc_ctas_renamed ORDER BY
id")
}
+ String v0_writer = "test_iceberg_jdbc_v0_writer"
+ String v0_reader = "test_iceberg_jdbc_v0_reader"
+ String v0_database = "iceberg_jdbc_v0_test"
+ String v0_uri =
"jdbc:postgresql://${externalEnvIp}:${jdbc_port}/${v0_database}"
+ def driverLoader = new URLClassLoader([new
File(local_driver_path).toURI().toURL()] as URL[],
+ this.class.classLoader)
+ def pgDriver =
driverLoader.loadClass("org.postgresql.Driver").getDeclaredConstructor().newInstance()
+ def pgProperties = new Properties()
+ pgProperties.setProperty("user", "postgres")
+ pgProperties.setProperty("password", "123456")
+ def pgConnection =
pgDriver.connect("jdbc:postgresql://${externalEnvIp}:${jdbc_port}/${pg_db}",
pgProperties)
try {
+ def statement = pgConnection.createStatement()
+ try {
+ // JDBC schema detection scans all PostgreSQL schemas, so isolate
V0 from the V1 database.
+ statement.execute("DROP DATABASE IF EXISTS ${v0_database} WITH
(FORCE)")
+ statement.execute("CREATE DATABASE ${v0_database}")
+ statement.execute("DROP ROLE IF EXISTS iceberg_jdbc_v0_reader")
+ statement.execute("CREATE ROLE iceberg_jdbc_v0_reader LOGIN
PASSWORD '123456'")
+ } finally {
+ statement.close()
+ }
+ pgConnection.close()
+ pgConnection = pgDriver.connect(v0_uri, pgProperties)
+ statement = pgConnection.createStatement()
+ try {
+ // Use the PyIceberg-compatible V0 layout, which has no
iceberg_type column.
+ statement.execute("""CREATE TABLE iceberg_tables (
+ catalog_name VARCHAR(255) NOT NULL, table_namespace
VARCHAR(255) NOT NULL,
+ table_name VARCHAR(255) NOT NULL, metadata_location
VARCHAR(1000),
+ previous_metadata_location VARCHAR(1000),
+ PRIMARY KEY (catalog_name, table_namespace, table_name))""")
+ statement.execute("""CREATE TABLE iceberg_namespace_properties (
+ catalog_name VARCHAR(255) NOT NULL, namespace VARCHAR(255) NOT
NULL,
+ property_key VARCHAR(255), property_value VARCHAR(1000),
+ PRIMARY KEY (catalog_name, namespace, property_key))""")
+ statement.execute("GRANT CONNECT ON DATABASE ${v0_database} TO
iceberg_jdbc_v0_reader")
+ statement.execute("GRANT USAGE ON SCHEMA public TO
iceberg_jdbc_v0_reader")
+ statement.execute("GRANT SELECT ON ALL TABLES IN SCHEMA public TO
iceberg_jdbc_v0_reader")
+ } finally {
+ statement.close()
+ }
+ } finally {
+ pgConnection.close()
+ driverLoader.close()
+ }
+
+ def createV0Catalog = { String name, String uri, String user ->
+ sql """DROP CATALOG IF EXISTS ${name}"""
+ sql """CREATE CATALOG ${name} PROPERTIES (
+ 'type' = 'iceberg', 'iceberg.catalog.type' = 'jdbc',
+ 'uri' = '${uri}', 'warehouse' = 's3://warehouse/jdbc_v0_wh/',
+ 'iceberg.jdbc.driver_url' = '${driver_name}',
+ 'iceberg.jdbc.driver_class' = 'org.postgresql.Driver',
+ 'iceberg.jdbc.user' = '${user}', 'iceberg.jdbc.password' =
'123456',
+ 'iceberg.jdbc.catalog_name' = 'jdbc_v0_catalog',
+ 'iceberg.jdbc.init-catalog-tables' = 'false',
'iceberg.jdbc.schema-version' = 'V0',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minio_port}',
+ 's3.access_key' = 'admin', 's3.secret_key' = 'password',
+ 's3.region' = 'us-east-1', 'use_path_style' = 'true')"""
+ }
+
+ try {
+ createV0Catalog(v0_writer, v0_uri, "postgres")
+ sql """CREATE DATABASE IF NOT EXISTS ${v0_writer}.v0_db"""
+ sql """CREATE TABLE ${v0_writer}.v0_db.v0_table (id INT) PROPERTIES
('format-version' = '2')"""
+ sql """INSERT INTO ${v0_writer}.v0_db.v0_table VALUES (1), (2)"""
+ // PostgreSQL enforces read-only transactions, so accidental catalog
migration or writes fail.
+ createV0Catalog(v0_reader, v0_uri +
"?options=-c%20default_transaction_read_only%3Don",
+ "iceberg_jdbc_v0_reader")
+ assertTrue(sql("SHOW DATABASES FROM ${v0_reader}").collect { it[0]
}.contains("v0_db"))
+ assertEquals([["v0_table"]], sql("SHOW TABLES FROM
${v0_reader}.v0_db"))
+ assertEquals("id", sql("DESC ${v0_reader}.v0_db.v0_table")[0][0])
+ assertEquals([[1], [2]], sql("SELECT * FROM
${v0_reader}.v0_db.v0_table ORDER BY id"))
+ def v0_snapshot = sql("SELECT snapshot_id FROM
${v0_reader}.v0_db.v0_table\$snapshots")[0][0]
+ assertEquals([[1], [2]], sql("""SELECT * FROM
${v0_reader}.v0_db.v0_table
+ FOR VERSION AS OF ${v0_snapshot} ORDER BY id"""))
+ sql """INSERT INTO ${v0_writer}.v0_db.v0_table VALUES (3)"""
+ sql """REFRESH CATALOG ${v0_reader}"""
+ assertEquals([[3L]], sql("SELECT COUNT(*) FROM
${v0_reader}.v0_db.v0_table"))
+ assertEquals([[2L]], sql("""SELECT COUNT(*) FROM
${v0_reader}.v0_db.v0_table
+ FOR VERSION AS OF ${v0_snapshot}"""))
// Clean up existing catalog
sql """DROP CATALOG IF EXISTS ${catalog_name}"""
@@ -540,6 +621,8 @@ suite("test_iceberg_jdbc_catalog", "p0,external") {
try {
sql """SWITCH internal"""
sql """DROP CATALOG IF EXISTS ${catalog_name}"""
+ sql """DROP CATALOG IF EXISTS ${v0_reader}"""
+ sql """DROP CATALOG IF EXISTS ${v0_writer}"""
} catch (Exception e) {
logger.warn("Failed to cleanup catalog: ${e.message}")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
