Skip to content

Flink: Support reading Iceberg views in SQL #17859

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
talatuyarer wants to merge 2 commits into
base: main
Choose a base branch
from
+372 −9
Open

Flink: Support reading Iceberg views in SQL #17859

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
beff342
Enhance FlinkCatalog to support view operations such as list, load
talatuyarer Aug 28, 2026
a68e7c7
Flink: Address review comments on view read path
talatuyarer Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 77 additions & 9 deletions flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.apache.flink.table.catalog.CatalogPartition;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.catalog.ResolvedSchema;
Expand Down Expand Up @@ -67,9 +68,11 @@
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.catalog.ViewCatalog;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchViewException;
import org.apache.iceberg.flink.util.FlinkAlterTableUtil;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
import org.apache.iceberg.io.CloseableIterable;
Expand All @@ -80,6 +83,9 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.view.SQLViewRepresentation;
import org.apache.iceberg.view.View;
import org.apache.iceberg.view.ViewProperties;

/**
* A Flink Catalog implementation that wraps an Iceberg {@link Catalog}.
Expand All @@ -94,10 +100,13 @@
*/
@Internal
public class FlinkCatalog extends AbstractCatalog {
private static final String FLINK_DIALECT = "flink";

private final CatalogLoader catalogLoader;
private final Catalog icebergCatalog;
private final Namespace baseNamespace;
private final SupportsNamespaces asNamespaceCatalog;
private final ViewCatalog asViewCatalog;
private final Closeable closeable;
private final boolean cacheEnabled;

Expand All @@ -120,6 +129,8 @@ public FlinkCatalog(
: originalCatalog;
asNamespaceCatalog =
originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;
// the caching wrapper only caches tables, so view operations always go to the original catalog
asViewCatalog = originalCatalog instanceof ViewCatalog ? (ViewCatalog) originalCatalog : null;
closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;

FlinkEnvironmentContext.init();
Expand Down Expand Up @@ -167,6 +178,12 @@ TableIdentifier toIdentifier(ObjectPath path) {
}
}

private boolean canBeView(ObjectPath tablePath) {
// the view catalog is only consulted for names that can denote a view: metadata-table
// syntax ("name$type") never does
return asViewCatalog != null && !tablePath.getObjectName().contains("$");
}

@Override
public List<String> listDatabases() throws CatalogException {
if (asNamespaceCatalog == null) {
Expand Down Expand Up @@ -321,8 +338,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
@Override
public List<String> listTables(String databaseName)
throws DatabaseNotExistException, CatalogException {
List<String> results;
try {
return icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream()
results =
icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream()
.map(TableIdentifier::name)
.collect(Collectors.toList());
} catch (NoSuchNamespaceException e) {
throw new DatabaseNotExistException(getName(), databaseName, e);
}

// Flink's Catalog#listTables contract requires this to return both tables and views
results.addAll(listViews(databaseName));
return results;
}

@Override
public List<String> listViews(String databaseName)
throws DatabaseNotExistException, CatalogException {
if (asViewCatalog == null) {
return Collections.emptyList();
}

try {
return asViewCatalog.listViews(appendLevel(baseNamespace, databaseName)).stream()
.map(TableIdentifier::name)
.collect(Collectors.toList());
} catch (NoSuchNamespaceException e) {
Expand All @@ -331,9 +370,24 @@ public List<String> listTables(String databaseName)
}

@Override
public CatalogTable getTable(ObjectPath tablePath)
public CatalogBaseTable getTable(ObjectPath tablePath)
throws TableNotExistException, CatalogException {
Table table = loadIcebergTable(tablePath);
Table table;
try {
table = loadIcebergTable(tablePath);
} catch (TableNotExistException e) {
if (!canBeView(tablePath)) {
throw e;
}

try {
View view = asViewCatalog.loadView(toIdentifier(tablePath));
return toCatalogView(view);
} catch (NoSuchViewException viewException) {
e.addSuppressed(viewException);
throw e;
}
}

// Flink's CREATE TABLE LIKE clause relies on properties sent back here to create new table.
// As Flink API accepts only Map<String, String> for props, here we are serializing catalog
Expand Down Expand Up @@ -378,7 +432,9 @@ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistExcepti

@Override
public boolean tableExists(ObjectPath tablePath) throws CatalogException {
return icebergCatalog.tableExists(toIdentifier(tablePath));
TableIdentifier identifier = toIdentifier(tablePath);
return icebergCatalog.tableExists(identifier)
|| (canBeView(tablePath) && asViewCatalog.viewExists(identifier));
}

@Override
Expand Down Expand Up @@ -683,6 +739,23 @@ static CatalogTable toCatalogTable(Table table) {
return toCatalogTableWithProps(table, table.properties());
}

private CatalogView toCatalogView(View view) {
SQLViewRepresentation sqlRepresentation = view.sqlFor(FLINK_DIALECT);
Preconditions.checkState(sqlRepresentation != null, "Cannot load SQL for view %s", view.name());

ResolvedSchema resolvedSchema = FlinkSchemaUtil.toResolvedSchema(view.schema());
org.apache.flink.table.api.Schema schema =
org.apache.flink.table.api.Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();

Map<String, String> options = Maps.newHashMap(view.properties());
String comment = options.remove(ViewProperties.COMMENT);

// both original and expanded query hold the stored SQL: Flink expands the query itself,
// resolving unqualified references against the view's own catalog and database
return CatalogView.of(
schema, comment, sqlRepresentation.sql(), sqlRepresentation.sql(), options);
Comment on lines +753 to +756

@singhpk234 singhpk234 Aug 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you please elaborate this a bit more, defaultCatalog / defaultNamespace should only be used for references inside the view sql ... for the whole view itself the current catalog and namespace should already have been resolved ?

@talatuyarer talatuyarer Sep 1, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct, they're resolution context for references inside the view SQL, not attributes of the view itself, and Flink's planner doesn't consume them. It resolves unqualified references against the view's own catalog and database. They were meant as informational output only, and the getName() fallback could even surface a value that isn't actually in the stored metadata. Removed both options; the CatalogView options
now carry only the Iceberg view properties.

}

@Override
public Optional<Factory> getFactory() {
return Optional.of(new FlinkDynamicTableFactory(this));
Expand All @@ -695,11 +768,6 @@ CatalogLoader getCatalogLoader() {
// ------------------------------ Unsupported methods
// ---------------------------------------------

@Override
public List<String> listViews(String databaseName) throws CatalogException {
return Collections.emptyList();
}

@Override
public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
throws CatalogException {
Expand Down
Loading
Loading