Skip to content

Flink: Support ALTER VIEW and add view-dialect-strict option in FlinkCatalog #17874

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 3 commits into
base: main
Choose a base branch
from
+727 −14
Open

Flink: Support ALTER VIEW and add view-dialect-strict option in FlinkCatalog #17874

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 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
5fb8064
Enhance FlinkCatalog to support altering views, including preserving …
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
259 changes: 246 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,11 @@
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.UpdateViewProperties;
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,12 +103,18 @@
*/
@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;
private final boolean viewDialectStrict;

public FlinkCatalog(
String catalogName,
Expand All @@ -108,10 +123,29 @@ public FlinkCatalog(
CatalogLoader catalogLoader,
boolean cacheEnabled,
long cacheExpirationIntervalMs) {
this(
catalogName,
defaultDatabase,
baseNamespace,
catalogLoader,
cacheEnabled,
cacheExpirationIntervalMs,
FlinkCatalogFactory.VIEW_DIALECT_STRICT_DEFAULT);
}

public FlinkCatalog(
String catalogName,
String defaultDatabase,
Namespace baseNamespace,
CatalogLoader catalogLoader,
boolean cacheEnabled,
long cacheExpirationIntervalMs,
boolean viewDialectStrict) {
super(catalogName, defaultDatabase);
this.catalogLoader = catalogLoader;
this.baseNamespace = baseNamespace;
this.cacheEnabled = cacheEnabled;
this.viewDialectStrict = viewDialectStrict;

Catalog originalCatalog = catalogLoader.loadCatalog();
icebergCatalog =
Expand All @@ -120,6 +154,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 +357,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
@Override
public List<String> listTables(String databaseName)
throws DatabaseNotExistException, CatalogException {
List<String> results;
try {
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 icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream()
return asViewCatalog.listViews(appendLevel(baseNamespace, databaseName)).stream()
.map(TableIdentifier::name)
.collect(Collectors.toList());
} catch (NoSuchNamespaceException e) {
Expand All @@ -331,9 +389,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("$")) {
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 +453,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 +487,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 +509,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 +584,85 @@ 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 void alterIcebergView(
ObjectPath tablePath, ResolvedCatalogView newView, boolean ignoreIfNotExists)
throws TableNotExistException, CatalogException {
View view;
try {
view = asViewCatalog.loadView(toIdentifier(tablePath));
} catch (NoSuchViewException e) {
if (!ignoreIfNotExists) {
throw new TableNotExistException(getName(), tablePath, e);
}

return;
}

SQLViewRepresentation currentRepresentation = view.sqlFor(FLINK_DIALECT);
if (currentRepresentation == null
|| !newView.getOriginalQuery().equals(currentRepresentation.sql())) {
// a new query becomes a new view version; the stored resolution context is preserved
view.replaceVersion()
.withQuery(FLINK_DIALECT, newView.getOriginalQuery())
.withSchema(FlinkSchemaUtil.convert(newView.getResolvedSchema()))
.withDefaultNamespace(view.currentVersion().defaultNamespace())
.commit();
}

// only set changed or added properties: Flink SQL cannot express a property removal
// (ALTER VIEW SET merges), and removing keys absent from the incoming options could
// strip properties that Flink did not carry over
Map<String, String> newProperties = Maps.newHashMap(newView.getOptions());
newProperties.remove(DEFAULT_CATALOG_OPTION);
newProperties.remove(DEFAULT_NAMESPACE_OPTION);
if (!StringUtils.isNullOrWhitespaceOnly(newView.getComment())) {
newProperties.put(ViewProperties.COMMENT, newView.getComment());
}

Map<String, String> currentProperties = view.properties();
UpdateViewProperties update = view.updateProperties();
boolean changed = false;
for (Map.Entry<String, String> entry : newProperties.entrySet()) {
if (!entry.getValue().equals(currentProperties.get(entry.getKey()))) {
update.set(entry.getKey(), entry.getValue());
changed = true;
}
}

if (changed) {
update.commit();
}
}

private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
if (!Objects.equals(ct1.getUnresolvedSchema(), ct2.getUnresolvedSchema())) {
throw new UnsupportedOperationException(
Expand Down Expand Up @@ -507,6 +697,20 @@ 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) {
if (asViewCatalog == null) {
throw new UnsupportedOperationException(
"Altering a view is not supported by catalog: " + getName());
}

Preconditions.checkArgument(
newTable instanceof ResolvedCatalogView,
"Expected a ResolvedCatalogView but got: %s",
newTable.getClass().getName());
alterIcebergView(tablePath, (ResolvedCatalogView) newTable, ignoreIfNotExists);
return;
}

validateFlinkTable(newTable);

Table icebergTable;
Expand Down Expand Up @@ -569,6 +773,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 +892,35 @@ 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());
if (viewDialectStrict && !FLINK_DIALECT.equalsIgnoreCase(sqlRepresentation.dialect())) {
throw new UnsupportedOperationException(
String.format(
"View %s does not have a flink dialect representation and %s is enabled",
view.name(), FlinkCatalogFactory.VIEW_DIALECT_STRICT));
}

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 +933,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