Skip to content

Flink: Support creating table and altering table in Flink SQL #1393

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

Merged
rdblue merged 7 commits into from
Sep 2, 2020
Merged

Flink: Support creating table and altering table in Flink SQL #1393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b003b29
Flink: Support creating table and altering table in Flink SQL
JingsongLi Aug 27, 2020
87f8c79
Fix case
JingsongLi Aug 27, 2020
5adac5c
Address comments
JingsongLi Aug 28, 2020
2d1aed6
checkstyle
JingsongLi Aug 28, 2020
6eb6a5c
Address comments
JingsongLi Aug 31, 2020
e1dcb7f
Address comment
JingsongLi Aug 31, 2020
ac06e52
checkstyles
JingsongLi Sep 1, 2020
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
200 changes: 185 additions & 15 deletions 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 @@ -24,6 +24,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.flink.table.api.TableSchema;
Expand All @@ -34,6 +35,7 @@
import org.apache.flink.table.catalog.CatalogFunction;
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.CatalogTableImpl;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
Expand All @@ -49,14 +51,22 @@
import org.apache.flink.util.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.CachingCatalog;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
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;

Expand Down Expand Up @@ -277,15 +287,14 @@ public List<String> listTables(String databaseName) throws DatabaseNotExistExcep
}

@Override
public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
try {
Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
Table table = loadIcebergTable(tablePath);
return toCatalogTable(table);
}

// NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
// catalog table.
// Let's re-loading table from Iceberg catalog when creating source/sink operators.
return new CatalogTableImpl(tableSchema, table.properties(), null);
private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistException {
try {
return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
Expand Down Expand Up @@ -320,19 +329,180 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}

/**
* TODO Add partitioning to the Flink DDL parser.
*/
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
throws CatalogException {
throw new UnsupportedOperationException("Not support createTable now.");
throws CatalogException, TableAlreadyExistException {
validateFlinkTable(table);

Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);

ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
String location = null;
for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
if ("location".equalsIgnoreCase(entry.getKey())) {
location = entry.getValue();
} else {
properties.put(entry.getKey(), entry.getValue());

@kbendick kbendick Aug 30, 2020

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.

Should location still be placed in the table properties or will that cause some kind of conflict / error?

@JingsongLi JingsongLi Aug 31, 2020

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.

I think there is no conflict/error, but I think it is good to reduce duplicate storage, cause iceberg has saved this information.

@rdblue rdblue Sep 2, 2020

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.

I'd prefer not to duplicate it in table properties. Then we would have to worry about keeping the two in sync.

}
}

try {
icebergCatalog.createTable(
toIdentifier(tablePath),
icebergSchema,
spec,
location,
properties.build());
} catch (AlreadyExistsException e) {
throw new TableAlreadyExistException(getName(), tablePath, e);
}
}

@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
throws CatalogException {
throw new UnsupportedOperationException("Not support alterTable now.");
throws CatalogException, TableNotExistException {
validateFlinkTable(newTable);
Table icebergTable = loadIcebergTable(tablePath);
CatalogTable table = toCatalogTable(icebergTable);

// Currently, Flink SQL only support altering table properties.

@openinx openinx Aug 27, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

What's the reason that we could not support adding /removing/renaming column ?

@JingsongLi JingsongLi Aug 27, 2020

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.

No Flink DLL to add/removing/renaming column...

@rdblue rdblue Aug 28, 2020

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.

I should also note that support for adding/removing/renaming columns cannot be done by comparing CatalogTable instances, unless the Flink schema contains Iceberg column IDs.

The problem is clear when you consider a simple example:

  • Iceberg table schema: id bigint, a float, b float
  • Flink table schema: id bigint, x float, y float

There are two ways to get the Flink schema: rename a -> x and b -> y, or drop a, drop b, add x, add y. Guessing which one was intended by the user is not okay because it would corrupt data. If the values from a are read when projecting x after a was actually dropped, then this is a serious correctness bug.

Also note that there are some transformations that can't be detected. For example, drop a then add a. The result should be that all values of column a are discarded. This happens when the wrong data was written to a column but the column is still needed for newer data.

@JingsongLi JingsongLi Aug 31, 2020 •
edited
Loading

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.

Good point.
If there is only one operation in a single call, it seems feasible. And actually, in the SQL DDLs, only a single type is in a single SQL.
But yes, this API is unclear, if we look at it from the API level alone, there are too many possibilities...
I'll add comments in the code.


// For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
// CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
if (!table.getSchema().equals(newTable.getSchema())) {
throw new UnsupportedOperationException("Altering schema is not supported yet.");
}

if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
}

Map<String, String> oldOptions = table.getOptions();
Map<String, String> setProperties = Maps.newHashMap();

String setLocation = null;
String setSnapshotId = null;
String pickSnapshotId = null;

for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
String key = entry.getKey();
String value = entry.getValue();

if (Objects.equals(value, oldOptions.get(key))) {
continue;
}

if ("location".equalsIgnoreCase(key)) {
setLocation = value;
} else if ("current-snapshot-id".equalsIgnoreCase(key)) {
setSnapshotId = value;
} else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
pickSnapshotId = value;
} else {
setProperties.put(key, value);
}
}

oldOptions.keySet().forEach(k -> {
if (!newTable.getOptions().containsKey(k)) {
setProperties.put(k, null);

@openinx openinx Aug 27, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Q: Does this align with the flink sql semantics ?
I saw the document said: "Set one or more properties in the specified table. If a particular property is already set in the table, override the old value with the new one."

ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...)

For the existing key-values (in old table ) which don't appear in the new table, should we remove them from old table ? ( The document did not describe this case clearly, just for confirmation).

@JingsongLi JingsongLi Aug 27, 2020

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.

You can take a look to tests.
The existing key-values will be keep. The newTable.getOptions() is not just from alter DDL. It is already merged with old options.
Actually, there is not DDL to delete key-value too...

@openinx openinx Aug 27, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

It is already merged with old options.

If the new options are already merged with old options, so for the key in old options, shouldn't it be always in new options ? Seems there's no reason to add this sentence here ?

@JingsongLi JingsongLi Aug 27, 2020

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.

What do you mean?
The new options is new all options.
For example:
old: {'j' = 'am', 'p' = 'an'}
alter: ALTER TABLE t UNSET TBLPROPERTIES ('j')
newTable.getOptions() will be: {'p' = 'an'}

@JingsongLi JingsongLi Aug 28, 2020

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.

I'll add a test for unsetting PROPERTIES.

@rdblue rdblue Aug 28, 2020

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.

I think it is okay to diff the property sets like this, but it seems like it would be easier not to. Right now, Flink has to apply the changes, then this code diffs the property sets, then Iceberg will re-apply the changes. In addition, this model doesn't work for schema updates, as I noted above.

@JingsongLi JingsongLi Aug 31, 2020 •
edited
Loading

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.

But properties updates does not have column IDs. As long as the last is the same.
Sorry, I don't get your point.

@openinx openinx Aug 31, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

setProperties.put(k, null) ? The javadoc from Map said :

 * @throws NullPointerException if the specified key or value is null
     *         and this map does not permit null keys or values
     * @throws IllegalArgumentException if some property of the specified key
     *         or value prevents it from being stored in this map
     */
    V put(K key, V value);

@rdblue rdblue Sep 2, 2020

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.

My point is that there isn't a correctness problem so this is okay. But, this causes Flink to do much more work because it has to apply changes from SQL, then recover those changes by comparing property maps, and pass the changes to Iceberg so that Iceberg can apply the changes. It is easier to pass the changes directly to Iceberg if the Flink API can be updated to support it.

@JingsongLi JingsongLi Sep 3, 2020

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.

Got it, I think we can have a try in Flink.

}
});

commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
}

private static void validateFlinkTable(CatalogBaseTable table) {
Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");

@rdblue rdblue Aug 28, 2020

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.

Is there a case where CatalogBaseTable doesn't implement CatalogTable?

@kbendick kbendick Aug 30, 2020

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.

CatalogTable is a subinterface that inherits from CatalogBaseTable. So definitely, yes.

See the java docs on the current CatalogBaseTable in Flink:
https://ci.apache.org/projects/flink/flink-docs-release-1.11/api/java/org/apache/flink/table/catalog/CatalogBaseTable.html

@JingsongLi JingsongLi Aug 31, 2020

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.

It is CatalogView, Iceberg catalog does not support views, so if there is a view, should be a bug...


TableSchema schema = table.getSchema();
schema.getTableColumns().forEach(column -> {
if (column.isGenerated()) {
throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
}
});

if (!schema.getWatermarkSpecs().isEmpty()) {
throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
}

if (schema.getPrimaryKey().isPresent()) {

@rdblue rdblue Aug 28, 2020

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.

Is this something we should add to Iceberg for Flink use cases? What does Flink use the primary key for?

@kbendick kbendick Aug 30, 2020

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.

I found this which might answer your question: https://cwiki.apache.org/confluence/display/FLINK/FLIP+87%3A+Primary+key+constraints+in+Table+API

In particular, here are the proposed changes:

Proposed Changes
We suggest to introduce the concept of primary key constraint as a hint for FLINK to leverage for optimizations.

Primary key constraints tell that a column or a set of columns of a table or a view are unique and they do not contain null.
Neither of columns in a primary can be nullable.
Primary key therefore uniquely identify a row in a table.

So it sounds just like an RDBMS primary key.

Note however, that even in the FLIP (which is just the proposal and not necessarily the finished product), it does state that there's no planned enforcement on the PK. It's up to the user to ensure that the PK is non-null and unique.

Primary key validity checks
SQL standard specifies that a constraint can either be ENFORCED or NOT ENFORCED.
This controls if the constraint checks are performed on the incoming/outgoing data.
Flink does not own the data therefore the only mode we want to support is the NOT ENFORCED mode.
Its up to the user to ensure that the query enforces key integrity.

So I agree here that throwing might be the most useful option and that there's likely nothing on the iceberg side to be added to enforce this as Flink doesn't enforce it either. In an entirely streaming setting, ensuring unique keys would be rather difficult and so to me it somewhat sounds like the PK is just more metadata that could very well be in TBLPROPERTIES.

But a more experienced Flink SQL user than myself might have more to say on the matter. I've never attempted to enforce a PK when using Flink SQL. Sounds like the work to do so would involve custom operators etc.

TLDR: The Primary Key is just a constraint, which is currently part of Flink's Table spec but goes unenforced and is up to the user. It does not appear as though the PK info is supported in any UpsertSinks etc, though that may be discussed / planned in the future. Support in the DDL for Primary Key constraints is relatively new (Flink 1.11 / current, with support in the API coming in at Flink 1.10).

@JingsongLi JingsongLi Aug 31, 2020 •
edited
Loading

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.

Thanks @kbendick .
At present, in Flink, PK is mainly used to process CDC stream.

  • For example, if user access a Kafka source of a CDC stream, the user can define a primary key. In this way, the downstream can perform efficient dynamic table / static table conversion (restore to the original static table) according to a certain primary key.
  • For example, when the stream data (CDC) is written into a JDBC database, the user can define primary key. In this way, Flink can insert the data into the database by using the upsert writing way.

I think, If iceberg supports CDC native processing in the future, we may be able to use it.

throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
}
}

@openinx openinx Aug 31, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Do we need to add a TODO indicating that we flink only support identity partition now but will support hidden column future ?

@JingsongLi JingsongLi Aug 31, 2020

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.

Already have todo in toPartitionKeys.

private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
partitionKeys.forEach(builder::identity);
return builder.build();
}

private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
List<String> partitionKeys = Lists.newArrayList();
for (PartitionField field : spec.fields()) {
if (field.transform().isIdentity()) {
partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
} else {
// Not created by Flink SQL.
// For compatibility with iceberg tables, return empty.
// TODO modify this after Flink support partition transform.
return Collections.emptyList();

@rdblue rdblue Aug 28, 2020

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.

All tables with any partition transform other than identity appear to be unpartitioned? Why not return all of the identity fields at least?

@kbendick kbendick Aug 30, 2020

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.

To me, it seems like adding all of the identity fields (but not the transformed fields) would likely be incorrect. Although returning an empty list when the table is partitioned seems like a possible correctness bug to me too.

Should we consider throwing an exception in this case instead until such a time that Flink supports partition transforms?

@JingsongLi JingsongLi Aug 31, 2020

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.

+1 to likely be incorrect.
But I don't want to throw an exception because it can read the existing iceberg table.
Actually, the returned partition keys are useless, except that Flink can show users the meta information of the table.

All partition operations are directly delegated to specific source / sink, so Flink does not need to see partition information.
I tend to support it so that we can read existing Iceberg tables.

@rdblue rdblue Sep 2, 2020

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.

I was thinking that since operations are delegated to Iceberg, correctness is not an issue. It would be nice to show users which columns are partition columns so they can see which ones are good candidates for query predicates. I don't think this is a blocker, though.

@JingsongLi JingsongLi Sep 3, 2020

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.

I see, maybe we can expose these information in properties.

}
}
return partitionKeys;
}

private static void commitChanges(Table table, String setLocation, String setSnapshotId,
String pickSnapshotId, Map<String, String> setProperties) {
// don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
// one order leads to different results
Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
"Cannot set the current snapshot ID and cherry-pick snapshot changes");

if (setSnapshotId != null) {
long newSnapshotId = Long.parseLong(setSnapshotId);
table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
}

// if updating the table snapshot, perform that update first in case it fails
if (pickSnapshotId != null) {
long newSnapshotId = Long.parseLong(pickSnapshotId);
table.manageSnapshots().cherrypick(newSnapshotId).commit();
}

Transaction transaction = table.newTransaction();

@rdblue rdblue Aug 28, 2020

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.

Should we do the operations above this point in the transaction as well? That seems reasonable to me. I'm not sure why we don't in other places.

@JingsongLi JingsongLi Aug 31, 2020

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.

Looks like the manageSnapshots is unsupported in TransactionTable.

@rdblue rdblue Sep 2, 2020

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.

That explains it. Thanks!


if (setLocation != null) {
transaction.updateLocation()
.setLocation(setLocation)
.commit();
}

if (!setProperties.isEmpty()) {
UpdateProperties updateProperties = transaction.updateProperties();
setProperties.forEach((k, v) -> {
if (v == null) {

@openinx openinx Aug 31, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

The v should never be null in HashMap ?

@JingsongLi JingsongLi Aug 31, 2020

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.

HashMap allows nulls:

 * Hash table based implementation of the <tt>Map</tt> interface.  This
 * implementation provides all of the optional map operations, and permits
 * <tt>null</tt> values and the <tt>null</tt> key.  (The <tt>HashMap</tt>
 * class is roughly equivalent to <tt>Hashtable</tt>, except that it is
 * unsynchronized and permits nulls.)

updateProperties.remove(k);
} else {
updateProperties.set(k, v);
}
});
updateProperties.commit();
}

transaction.commitTransaction();
}

static CatalogTable toCatalogTable(Table table) {
TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());

// NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer may use
// CatalogTableImpl to copy a new catalog table.
// Let's re-loading table from Iceberg catalog when creating source/sink operators.
// Iceberg does not have Table comment, so pass a null (Default comment value in Flink).
return new CatalogTableImpl(schema, partitionKeys, table.properties(), null);
}

// ------------------------------ Unsupported methods ---------------------------------------------
Expand Down
Loading