This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new d2c49d48a8 [#11130] Fix(flink-connector): Treat ForbiddenException as
table-not-exist during speculative table lookup (#11286)
d2c49d48a8 is described below
commit d2c49d48a8e33090383adac3f3cdf8f62734f4dc
Author: geyanggang <[email protected]>
AuthorDate: Fri May 29 11:25:59 2026 +0800
[#11130] Fix(flink-connector): Treat ForbiddenException as table-not-exist
during speculative table lookup (#11286)
### What changes were proposed in this pull request?
Catch `ForbiddenException` in the Flink connector's table lookup methods
and treat it as "table not exist":
- `BaseCatalog.getTable()` — throw `TableNotExistException` on
`ForbiddenException`
- `BaseCatalog.tableExists()` — return `false` on `ForbiddenException`
- `GravitinoHiveCatalog.getTable()` — throw `TableNotExistException` on
`ForbiddenException`
### Why are the changes needed?
When Flink Calcite resolves two-part SQL identifiers (e.g.,
`schema.table`), it speculatively calls `loadTable` with different
namespace combinations. If the probed table does not exist, the
authorization interceptor returns 403 before the method can return 404.
This blocks Calcite's fallback resolution and causes SQL validation
failures.
Authorization should not apply to non-existent resources. Returning 403
for a non-existent table is misleading and breaks Flink Calcite's
standard identifier resolution.
Fix: #11130
### Does this PR introduce _any_ user-facing change?
No API changes. Users with authorization enabled will no longer see
`ForbiddenException` when Flink Calcite speculatively probes
non-existent tables during SQL parsing.
### How was this patch tested?
Existing unit tests pass. The fix was verified in a production
environment with authorization enabled — two-part identifier queries
(e.g., `SELECT * FROM schema.table`) now resolve correctly without
requiring three-part references.
---
.../org/apache/gravitino/flink/connector/catalog/BaseCatalog.java | 8 ++++++++
.../gravitino/flink/connector/hive/GravitinoHiveCatalog.java | 3 +++
2 files changed, 11 insertions(+)
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
index f3fa79c29d..0d781fb631 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
@@ -68,6 +68,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.Schema;
import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTableException;
@@ -255,6 +256,11 @@ public abstract class BaseCatalog extends AbstractCatalog {
return toFlinkTable(table, tablePath);
} catch (NoSuchTableException e) {
// Fall through to check views.
+ } catch (ForbiddenException e) {
+ // Flink/Calcite speculatively probes tables during multi-part
identifier resolution.
+ // Treat authorization failure as table-not-exist to allow Calcite to
fall back to
+ // alternative resolution paths (e.g., treating the name as a schema).
+ throw new TableNotExistException(catalogName(), tablePath, e);
} catch (Exception e) {
LOG.warn("Failed to load table {} from catalog {}", ident,
catalogName(), e);
throw new CatalogException(e);
@@ -271,6 +277,8 @@ public abstract class BaseCatalog extends AbstractCatalog {
if (catalog().asTableCatalog().tableExists(ident)) {
return true;
}
+ } catch (ForbiddenException e) {
+ return false;
} catch (Exception e) {
throw new CatalogException(e);
}
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
index a027b36dd4..83d2acb11b 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
@@ -41,6 +41,7 @@ import
org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import org.apache.flink.table.factories.Factory;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTableException;
import org.apache.gravitino.exceptions.TableAlreadyExistsException;
@@ -160,6 +161,8 @@ public class GravitinoHiveCatalog extends BaseCatalog {
return super.toFlinkTable(table, tablePath);
} catch (NoSuchTableException e) {
// Fall through to check views.
+ } catch (ForbiddenException e) {
+ throw new TableNotExistException(catalogName(), tablePath, e);
} catch (Exception e) {
LOG.warn("Failed to load table {} from catalog {}", tablePath,
catalogName(), e);
throw new CatalogException(e);
