Skip to content

Flink: Support CREATE VIEW, DROP VIEW and ALTER VIEW RENAME in FlinkCatalog #17873

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
+546 −13
Open

Flink: Support CREATE VIEW, DROP VIEW and ALTER VIEW RENAME in FlinkCatalog #17873

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
92506f6
Enhance FlinkCatalog to support creating, dropping, and renaming view…
talatuyarer Aug 29, 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
173 changes: 160 additions & 13 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,8 +36,10 @@
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.ResolvedCatalogView;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
import org.apache.flink.table.catalog.exceptions.CatalogException;
Expand Down Expand Up @@ -67,9 +69,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 +84,10 @@
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;
import org.apache.iceberg.view.ViewVersion;

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

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 +133,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 @@ -321,8 +336,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 +368,26 @@ 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) {
// metadata tables ("name$type") can never be views, and without a view catalog there is
// nothing else to look up
if (asViewCatalog == null || tablePath.getObjectName().contains("$")) {

@pvary pvary 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.

This check could be a method, and could be reused in exists, drop too

@pvary pvary 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.

Or just ignore this check here too, and try anyway... My main point it is not consistent

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,18 +432,29 @@ 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)
|| (asViewCatalog != null && asViewCatalog.viewExists(identifier));
}

@Override
public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
TableIdentifier identifier = toIdentifier(tablePath);

boolean dropped;
try {
icebergCatalog.dropTable(toIdentifier(tablePath));
dropped = icebergCatalog.dropTable(identifier);
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
if (!ignoreIfNotExists) {
throw new TableNotExistException(getName(), tablePath, e);
}
dropped = false;
}

if (!dropped && asViewCatalog != null) {
dropped = asViewCatalog.dropView(identifier);
}

if (!dropped && !ignoreIfNotExists) {
throw new TableNotExistException(getName(), tablePath);
}
}

Expand All @@ -401,6 +466,17 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
toIdentifier(tablePath),
toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName)));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
if (asViewCatalog != null) {
try {
asViewCatalog.renameView(
toIdentifier(tablePath),
toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName)));
return;
} catch (NoSuchViewException viewException) {
e.addSuppressed(viewException);
}
}

if (!ignoreIfNotExists) {
throw new TableNotExistException(getName(), tablePath, e);
}
Expand All @@ -412,6 +488,20 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
throws CatalogException, TableAlreadyExistException {
if (table instanceof CatalogView) {
if (asViewCatalog == null) {
throw new UnsupportedOperationException(
"Creating a view is not supported by catalog: " + getName());
}

Preconditions.checkArgument(
table instanceof ResolvedCatalogView,
"Expected a ResolvedCatalogView but got: %s",
table.getClass().getName());
createIcebergView(tablePath, (ResolvedCatalogView) table, ignoreIfExists);
return;
}

// Creating Iceberg table using connector is allowed only when table is created using LIKE
if (Objects.equals(
table.getOptions().get(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY),
Expand Down Expand Up @@ -473,6 +563,35 @@ private boolean isReservedProperty(String prop) {
|| FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(prop);
}

private void createIcebergView(
ObjectPath tablePath, ResolvedCatalogView view, boolean ignoreIfExists)
throws CatalogException, TableAlreadyExistException {
Map<String, String> properties = Maps.newHashMap(view.getOptions());
properties.remove(DEFAULT_CATALOG_OPTION);
properties.remove(DEFAULT_NAMESPACE_OPTION);
if (!StringUtils.isNullOrWhitespaceOnly(view.getComment())) {
properties.put(ViewProperties.COMMENT, view.getComment());
}

try {
// the default catalog is intentionally left unset so that readers resolve it to the
// name this catalog is registered under (see toCatalogView)
asViewCatalog
.buildView(toIdentifier(tablePath))
.withSchema(FlinkSchemaUtil.convert(view.getResolvedSchema()))
.withDefaultNamespace(appendLevel(baseNamespace, tablePath.getDatabaseName()))
.withQuery(FLINK_DIALECT, view.getOriginalQuery())
.withProperties(properties)
.create();
} catch (AlreadyExistsException e) {
if (!ignoreIfExists) {
throw new TableAlreadyExistException(getName(), tablePath, e);
}
} catch (NoSuchNamespaceException e) {
throw new CatalogException("Database does not exist: " + tablePath.getDatabaseName(), e);
}
}

private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
if (!Objects.equals(ct1.getUnresolvedSchema(), ct2.getUnresolvedSchema())) {
throw new UnsupportedOperationException(
Expand Down Expand Up @@ -507,6 +626,11 @@ private static void validateTablePartition(CatalogTable ct1, CatalogTable ct2) {
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
throws CatalogException, TableNotExistException {
if (newTable instanceof CatalogView) {
throw new UnsupportedOperationException(
"Altering a view is not supported yet for catalog: " + getName());
}

validateFlinkTable(newTable);

Table icebergTable;
Expand Down Expand Up @@ -569,6 +693,11 @@ public void alterTable(
List<TableChange> tableChanges,
boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
if (newTable instanceof CatalogView) {
throw new UnsupportedOperationException(
"Altering a view is not supported yet for catalog: " + getName());
}

validateFlinkTable(newTable);

Table icebergTable;
Expand Down Expand Up @@ -683,6 +812,29 @@ 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);

ViewVersion currentVersion = view.currentVersion();
String defaultCatalog =
currentVersion.defaultCatalog() != null ? currentVersion.defaultCatalog() : getName();
options.put(DEFAULT_CATALOG_OPTION, defaultCatalog);
options.put(DEFAULT_NAMESPACE_OPTION, currentVersion.defaultNamespace().toString());

// 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);
}

@Override
public Optional<Factory> getFactory() {
return Optional.of(new FlinkDynamicTableFactory(this));
Expand All @@ -695,11 +847,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