properties, Configuration conf, String impl) {
+ this.hadoopConf = new SerializableConfiguration(conf);
+ this.properties = Maps.newHashMap(properties); // wrap into a hashmap for serialization
+ this.name = name;
+ this.impl =
+ Preconditions.checkNotNull(
+ impl, "Cannot initialize custom Catalog, impl class name is null");
+ }
+
+ @Override
+ public Catalog loadCatalog() {
+ return CatalogUtil.loadCatalog(impl, name, properties, hadoopConf.get());
+ }
+
+ @Override
+ @SuppressWarnings({"checkstyle:NoClone", "checkstyle:SuperClone"})
+ public CatalogLoader clone() {
+ return new CustomCatalogLoader(name, properties, new Configuration(hadoopConf.get()), impl);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this).add("name", name).add("impl", impl).toString();
+ }
+ }
+}
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
new file mode 100644
index 000000000000..1d1505a28c05
--- /dev/null
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -0,0 +1,879 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+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.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.util.StringUtils;
+import org.apache.iceberg.CachingCatalog;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+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.flink.util.FlinkAlterTableUtil;
+import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+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;
+
+/**
+ * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}.
+ *
+ * The mapping between Flink database and Iceberg namespace: Supplying a base namespace for a
+ * given catalog, so if you have a catalog that supports a 2-level namespace, you would supply the
+ * first level in the catalog configuration and the second level would be exposed as Flink
+ * databases.
+ *
+ *
The Iceberg table manages its partitions by itself. The partition of the Iceberg table is
+ * independent of the partition of Flink.
+ */
+@Internal
+public class FlinkCatalog extends AbstractCatalog {
+ private final CatalogLoader catalogLoader;
+ private final Catalog icebergCatalog;
+ private final Namespace baseNamespace;
+ private final SupportsNamespaces asNamespaceCatalog;
+ private final Closeable closeable;
+ private final boolean cacheEnabled;
+
+ public FlinkCatalog(
+ String catalogName,
+ String defaultDatabase,
+ Namespace baseNamespace,
+ CatalogLoader catalogLoader,
+ boolean cacheEnabled,
+ long cacheExpirationIntervalMs) {
+ super(catalogName, defaultDatabase);
+ this.catalogLoader = catalogLoader;
+ this.baseNamespace = baseNamespace;
+ this.cacheEnabled = cacheEnabled;
+
+ Catalog originalCatalog = catalogLoader.loadCatalog();
+ icebergCatalog =
+ cacheEnabled
+ ? CachingCatalog.wrap(originalCatalog, cacheExpirationIntervalMs)
+ : originalCatalog;
+ asNamespaceCatalog =
+ originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;
+ closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;
+
+ FlinkEnvironmentContext.init();
+ }
+
+ @Override
+ public void open() throws CatalogException {}
+
+ @Override
+ public void close() throws CatalogException {
+ if (closeable != null) {
+ try {
+ closeable.close();
+ } catch (IOException e) {
+ throw new CatalogException(e);
+ }
+ }
+ }
+
+ public Catalog catalog() {
+ return icebergCatalog;
+ }
+
+ /** Append a new level to the base namespace */
+ private static Namespace appendLevel(Namespace baseNamespace, String newLevel) {
+ String[] namespace = new String[baseNamespace.levels().length + 1];
+ System.arraycopy(baseNamespace.levels(), 0, namespace, 0, baseNamespace.levels().length);
+ namespace[baseNamespace.levels().length] = newLevel;
+ return Namespace.of(namespace);
+ }
+
+ TableIdentifier toIdentifier(ObjectPath path) {
+ String objectName = path.getObjectName();
+ List tableName = Splitter.on('$').splitToList(objectName);
+
+ if (tableName.size() == 1) {
+ return TableIdentifier.of(
+ appendLevel(baseNamespace, path.getDatabaseName()), path.getObjectName());
+ } else if (tableName.size() == 2 && MetadataTableType.from(tableName.get(1)) != null) {
+ return TableIdentifier.of(
+ appendLevel(appendLevel(baseNamespace, path.getDatabaseName()), tableName.get(0)),
+ tableName.get(1));
+ } else {
+ throw new IllegalArgumentException("Illegal table name:" + objectName);
+ }
+ }
+
+ @Override
+ public List listDatabases() throws CatalogException {
+ if (asNamespaceCatalog == null) {
+ return Collections.singletonList(getDefaultDatabase());
+ }
+
+ return asNamespaceCatalog.listNamespaces(baseNamespace).stream()
+ .map(n -> n.level(n.levels().length - 1))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (asNamespaceCatalog == null) {
+ if (!getDefaultDatabase().equals(databaseName)) {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ } else {
+ return new CatalogDatabaseImpl(Maps.newHashMap(), "");
+ }
+ } else {
+ try {
+ Map metadata =
+ Maps.newHashMap(
+ asNamespaceCatalog.loadNamespaceMetadata(appendLevel(baseNamespace, databaseName)));
+ String comment = metadata.remove("comment");
+ return new CatalogDatabaseImpl(metadata, comment);
+ } catch (NoSuchNamespaceException e) {
+ throw new DatabaseNotExistException(getName(), databaseName, e);
+ }
+ }
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) throws CatalogException {
+ try {
+ getDatabase(databaseName);
+ return true;
+ } catch (DatabaseNotExistException ignore) {
+ return false;
+ }
+ }
+
+ @Override
+ public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ createDatabase(
+ name, mergeComment(database.getProperties(), database.getComment()), ignoreIfExists);
+ }
+
+ private void createDatabase(
+ String databaseName, Map metadata, boolean ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ if (asNamespaceCatalog != null) {
+ try {
+ asNamespaceCatalog.createNamespace(appendLevel(baseNamespace, databaseName), metadata);
+ } catch (AlreadyExistsException e) {
+ if (!ignoreIfExists) {
+ throw new DatabaseAlreadyExistException(getName(), databaseName, e);
+ }
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ "Namespaces are not supported by catalog: " + getName());
+ }
+ }
+
+ private Map mergeComment(Map metadata, String comment) {
+ Map ret = Maps.newHashMap(metadata);
+ if (metadata.containsKey("comment")) {
+ throw new CatalogException("Database properties should not contain key: 'comment'.");
+ }
+
+ if (!StringUtils.isNullOrWhitespaceOnly(comment)) {
+ ret.put("comment", comment);
+ }
+ return ret;
+ }
+
+ @Override
+ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade)
+ throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException {
+ if (asNamespaceCatalog != null) {
+ try {
+ boolean success = asNamespaceCatalog.dropNamespace(appendLevel(baseNamespace, name));
+ if (!success && !ignoreIfNotExists) {
+ throw new DatabaseNotExistException(getName(), name);
+ }
+ } catch (NoSuchNamespaceException e) {
+ if (!ignoreIfNotExists) {
+ throw new DatabaseNotExistException(getName(), name, e);
+ }
+ } catch (NamespaceNotEmptyException e) {
+ throw new DatabaseNotEmptyException(getName(), name, e);
+ }
+ } else {
+ if (!ignoreIfNotExists) {
+ throw new DatabaseNotExistException(getName(), name);
+ }
+ }
+ }
+
+ @Override
+ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists)
+ throws DatabaseNotExistException, CatalogException {
+ if (asNamespaceCatalog != null) {
+ Namespace namespace = appendLevel(baseNamespace, name);
+ Map updates = Maps.newHashMap();
+ Set removals = Sets.newHashSet();
+
+ try {
+ Map oldProperties = asNamespaceCatalog.loadNamespaceMetadata(namespace);
+ Map newProperties =
+ mergeComment(newDatabase.getProperties(), newDatabase.getComment());
+
+ for (String key : oldProperties.keySet()) {
+ if (!newProperties.containsKey(key)) {
+ removals.add(key);
+ }
+ }
+
+ for (Map.Entry entry : newProperties.entrySet()) {
+ if (!entry.getValue().equals(oldProperties.get(entry.getKey()))) {
+ updates.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ if (!updates.isEmpty()) {
+ asNamespaceCatalog.setProperties(namespace, updates);
+ }
+
+ if (!removals.isEmpty()) {
+ asNamespaceCatalog.removeProperties(namespace, removals);
+ }
+
+ } catch (NoSuchNamespaceException e) {
+ if (!ignoreIfNotExists) {
+ throw new DatabaseNotExistException(getName(), name, e);
+ }
+ }
+ } else {
+ if (getDefaultDatabase().equals(name)) {
+ throw new CatalogException(
+ "Can not alter the default database when the iceberg catalog doesn't support namespaces.");
+ }
+ if (!ignoreIfNotExists) {
+ throw new DatabaseNotExistException(getName(), name);
+ }
+ }
+ }
+
+ @Override
+ public List listTables(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ try {
+ return icebergCatalog.listTables(appendLevel(baseNamespace, databaseName)).stream()
+ .map(TableIdentifier::name)
+ .collect(Collectors.toList());
+ } catch (NoSuchNamespaceException e) {
+ throw new DatabaseNotExistException(getName(), databaseName, e);
+ }
+ }
+
+ @Override
+ public CatalogTable getTable(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ Table table = loadIcebergTable(tablePath);
+
+ // Flink's CREATE TABLE LIKE clause relies on properties sent back here to create new table.
+ // As Flink API accepts only Map for props, here we are serializing catalog
+ // name, database, table as json string to distinguish between catalog info
+ // and table properties in createTable.
+ String srcCatalogProps =
+ FlinkCreateTableOptions.toJson(
+ getName(), tablePath.getDatabaseName(), tablePath.getObjectName());
+
+ Map tableProps = table.properties();
+ if (tableProps.containsKey(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY)
+ || tableProps.containsKey(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Source table %s contains one/all of the reserved property keys: %s, %s.",
+ tablePath,
+ FlinkCreateTableOptions.CONNECTOR_PROPS_KEY,
+ FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY));
+ }
+
+ ImmutableMap.Builder mergedProps = ImmutableMap.builder();
+ mergedProps.put(
+ FlinkCreateTableOptions.CONNECTOR_PROPS_KEY, FlinkDynamicTableFactory.FACTORY_IDENTIFIER);
+ mergedProps.put(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY, srcCatalogProps);
+ mergedProps.putAll(tableProps);
+
+ return toCatalogTableWithProps(table, mergedProps.build());
+ }
+
+ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
+ if (cacheEnabled) {
+ table.refresh();
+ }
+
+ return table;
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ throw new TableNotExistException(getName(), tablePath, e);
+ }
+ }
+
+ @Override
+ public boolean tableExists(ObjectPath tablePath) throws CatalogException {
+ return icebergCatalog.tableExists(toIdentifier(tablePath));
+ }
+
+ @Override
+ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ try {
+ icebergCatalog.dropTable(toIdentifier(tablePath));
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ if (!ignoreIfNotExists) {
+ throw new TableNotExistException(getName(), tablePath, e);
+ }
+ }
+ }
+
+ @Override
+ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists)
+ throws TableNotExistException, TableAlreadyExistException, CatalogException {
+ try {
+ icebergCatalog.renameTable(
+ toIdentifier(tablePath),
+ toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName)));
+ } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
+ if (!ignoreIfNotExists) {
+ throw new TableNotExistException(getName(), tablePath, e);
+ }
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
+ }
+
+ @Override
+ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
+ throws CatalogException, TableAlreadyExistException {
+ // Creating Iceberg table using connector is allowed only when table is created using LIKE
+ if (Objects.equals(
+ table.getOptions().get(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY),
+ FlinkDynamicTableFactory.FACTORY_IDENTIFIER)
+ && table.getOptions().get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY) == null) {
+ throw new IllegalArgumentException(
+ "Cannot create the table with 'connector'='iceberg' table property in "
+ + "an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or "
+ + "create table without 'connector'='iceberg' related properties in an iceberg table.");
+ }
+
+ Preconditions.checkArgument(
+ table instanceof ResolvedCatalogTable,
+ "Expected a ResolvedCatalogTable but got: %s. "
+ + "Iceberg Flink catalog only supports resolved catalog tables "
+ + "(Materialized tables and other table kinds are not supported).",
+ table == null ? "null" : table.getClass().getName());
+ createIcebergTable(tablePath, (ResolvedCatalogTable) table, ignoreIfExists);
+ }
+
+ void createIcebergTable(ObjectPath tablePath, ResolvedCatalogTable table, boolean ignoreIfExists)
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getResolvedSchema());
+ PartitionSpec spec = toPartitionSpec(table.getPartitionKeys(), icebergSchema);
+ ImmutableMap.Builder properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry entry : table.getOptions().entrySet()) {
+ if (!isReservedProperty(entry.getKey())) {
+ properties.put(entry.getKey(), entry.getValue());
+ } else {
+ // Filtering reserved properties like catalog properties(added to support CREATE TABLE LIKE
+ // in getTable()), location and not persisting on table properties.
+ if (FlinkCreateTableOptions.LOCATION_KEY.equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ }
+ }
+ }
+
+ String comment = table.getComment();
+ if (comment != null && !comment.isEmpty()) {
+ properties.put(TableProperties.COMMENT, comment);
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath), icebergSchema, spec, location, properties.build());
+ } catch (AlreadyExistsException e) {
+ if (!ignoreIfExists) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
+ }
+ }
+
+ private boolean isReservedProperty(String prop) {
+ return FlinkCreateTableOptions.LOCATION_KEY.equalsIgnoreCase(prop)
+ || FlinkCreateTableOptions.CONNECTOR_PROPS_KEY.equalsIgnoreCase(prop)
+ || FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(prop);
+ }
+
+ private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
+ if (!Objects.equals(ct1.getUnresolvedSchema(), ct2.getUnresolvedSchema())) {
+ throw new UnsupportedOperationException(
+ "Altering schema is not supported in the old alterTable API. "
+ + "To alter schema, use the other alterTable API and provide a list of TableChange's.");
+ }
+
+ validateTablePartition(ct1, ct2);
+ }
+
+ private static void validateTablePartition(CatalogTable ct1, CatalogTable ct2) {
+ if (!ct1.getPartitionKeys().equals(ct2.getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+ }
+
+ /**
+ * This alterTable API only supports altering table properties.
+ *
+ * Support for adding/removing/renaming columns cannot be done by comparing CatalogTable
+ * instances, unless the Flink schema contains Iceberg column IDs.
+ *
+ *
To alter columns, use the other alterTable API and provide a list of TableChange's.
+ *
+ * @param tablePath path of the table or view to be modified
+ * @param newTable the new table definition
+ * @param ignoreIfNotExists flag to specify behavior when the table or view does not exist: if set
+ * to false, throw an exception, if set to true, do nothing.
+ * @throws CatalogException in case of any runtime exception
+ * @throws TableNotExistException if the table does not exist
+ */
+ @Override
+ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ CatalogTable table = toCatalogTable(icebergTable);
+ validateTableSchemaAndPartition(table, (CatalogTable) newTable);
+
+ Map oldProperties = table.getOptions();
+ Map setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldProperties.get(key))) {
+ continue;
+ }
+
+ if (FlinkCreateTableOptions.LOCATION_KEY.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);
+ }
+ }
+
+ oldProperties
+ .keySet()
+ .forEach(
+ k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ FlinkAlterTableUtil.commitChanges(
+ icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ validateFlinkTable(newTable);
+
+ Table icebergTable;
+ try {
+ icebergTable = loadIcebergTable(tablePath);
+ } catch (TableNotExistException e) {
+ if (!ignoreIfNotExists) {
+ throw e;
+ } else {
+ return;
+ }
+ }
+
+ // Does not support altering partition yet.
+ validateTablePartition(toCatalogTable(icebergTable), (CatalogTable) newTable);
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String cherrypickSnapshotId = null;
+
+ List propertyChanges = Lists.newArrayList();
+ List schemaChanges = Lists.newArrayList();
+ for (TableChange change : tableChanges) {
+ if (change instanceof TableChange.SetOption) {
+ TableChange.SetOption set = (TableChange.SetOption) change;
+
+ if (FlinkCreateTableOptions.LOCATION_KEY.equalsIgnoreCase(set.getKey())) {
+ setLocation = set.getValue();
+ } else if ("current-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ setSnapshotId = set.getValue();
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(set.getKey())) {
+ cherrypickSnapshotId = set.getValue();
+ } else {
+ propertyChanges.add(change);
+ }
+ } else if (change instanceof TableChange.ResetOption) {
+ propertyChanges.add(change);
+ } else {
+ schemaChanges.add(change);
+ }
+ }
+
+ FlinkAlterTableUtil.commitChanges(
+ icebergTable,
+ setLocation,
+ setSnapshotId,
+ cherrypickSnapshotId,
+ schemaChanges,
+ propertyChanges);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(
+ table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ org.apache.flink.table.api.Schema schema = table.getUnresolvedSchema();
+ schema
+ .getColumns()
+ .forEach(
+ column -> {
+ if (!FlinkCompatibilityUtil.isPhysicalColumn(column)) {
+ 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.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ ImmutableList.Builder partitionKeysBuilder = ImmutableList.builder();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeysBuilder.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();
+ }
+ }
+ return partitionKeysBuilder.build();
+ }
+
+ static CatalogTable toCatalogTableWithProps(Table table, Map props) {
+ ResolvedSchema resolvedSchema = FlinkSchemaUtil.toResolvedSchema(table.schema());
+ List partitionKeys = toPartitionKeys(table.spec(), table.schema());
+
+ // NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer
+ // may use DefaultCatalogTable to copy a new catalog table.
+ // Let's re-loading table from Iceberg catalog when creating source/sink operators.
+ return CatalogTable.newBuilder()
+ .schema(
+ org.apache.flink.table.api.Schema.newBuilder()
+ .fromResolvedSchema(resolvedSchema)
+ .build())
+ .partitionKeys(partitionKeys)
+ .options(props)
+ .build();
+ }
+
+ static CatalogTable toCatalogTable(Table table) {
+ return toCatalogTableWithProps(table, table.properties());
+ }
+
+ @Override
+ public Optional getFactory() {
+ return Optional.of(new FlinkDynamicTableFactory(this));
+ }
+
+ CatalogLoader getCatalogLoader() {
+ return catalogLoader;
+ }
+
+ // ------------------------------ Unsupported methods
+ // ---------------------------------------------
+
+ @Override
+ public List listViews(String databaseName) throws CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void createPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition partition,
+ boolean ignoreIfExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropPartition(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition newPartition,
+ boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listFunctions(String dbName) throws CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogFunction getFunction(ObjectPath functionPath)
+ throws FunctionNotExistException, CatalogException {
+ throw new FunctionNotExistException(getName(), functionPath);
+ }
+
+ @Override
+ public boolean functionExists(ObjectPath functionPath) throws CatalogException {
+ return false;
+ }
+
+ @Override
+ public void createFunction(
+ ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterFunction(
+ ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterTableStatistics(
+ ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterTableColumnStatistics(
+ ObjectPath tablePath, CatalogColumnStatistics columnStatistics, boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartitionStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogTableStatistics partitionStatistics,
+ boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartitionColumnStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogColumnStatistics columnStatistics,
+ boolean ignoreIfNotExists)
+ throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitions(ObjectPath tablePath)
+ throws TableNotExistException, TableNotPartitionedException, CatalogException {
+ Table table = loadIcebergTable(tablePath);
+
+ if (table.spec().isUnpartitioned()) {
+ throw new TableNotPartitionedException(icebergCatalog.name(), tablePath);
+ }
+
+ Set set = Sets.newHashSet();
+ try (CloseableIterable tasks = table.newScan().planFiles()) {
+ for (DataFile dataFile : CloseableIterable.transform(tasks, FileScanTask::file)) {
+ Map map = Maps.newHashMap();
+ StructLike structLike = dataFile.partition();
+ PartitionSpec spec = table.specs().get(dataFile.specId());
+ for (int i = 0; i < structLike.size(); i++) {
+ map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class)));
+ }
+ set.add(new CatalogPartitionSpec(map));
+ }
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Failed to list partitions of table %s", tablePath), e);
+ }
+
+ return Lists.newArrayList(set);
+ }
+
+ @Override
+ public List listPartitions(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitionsByFilter(
+ ObjectPath tablePath, List filters) throws CatalogException {
+ throw new UnsupportedOperationException();
+ }
+
+ // After partition pruning and filter push down, the statistics have become very inaccurate, so
+ // the statistics from
+ // here are of little significance.
+ // Flink will support something like SupportsReportStatistics in future.
+
+ @Override
+ public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) throws CatalogException {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath)
+ throws CatalogException {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogTableStatistics getPartitionStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getPartitionColumnStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+}
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
new file mode 100644
index 000000000000..24e2bdbba37a
--- /dev/null
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.util.PropertyUtil;
+
+/**
+ * A Flink Catalog factory implementation that creates {@link FlinkCatalog}.
+ *
+ * This supports the following catalog configuration options:
+ *
+ *
+ * type - Flink catalog factory key, should be "iceberg"
+ * catalog-type - iceberg catalog type, "hive", "hadoop" or "rest"
+ * uri - the Hive Metastore URI (Hive catalog only)
+ * clients - the Hive Client Pool Size (Hive catalog only)
+ * warehouse - the warehouse path (Hadoop catalog only)
+ * default-database - a database name to use as the default
+ * base-namespace - a base namespace as the prefix for all databases (Hadoop
+ * catalog only)
+ * cache-enabled - whether to enable catalog cache
+ *
+ *
+ * To use a custom catalog that is not a Hive or Hadoop catalog, extend this class and override
+ * {@link #createCatalogLoader(String, Map, Configuration)}.
+ */
+public class FlinkCatalogFactory implements CatalogFactory {
+
+ public static final String FACTORY_IDENTIFIER = "iceberg";
+
+ // Can not just use "type", it conflicts with CATALOG_TYPE.
+ public static final String ICEBERG_CATALOG_TYPE = "catalog-type";
+ public static final String ICEBERG_CATALOG_TYPE_HADOOP = "hadoop";
+ public static final String ICEBERG_CATALOG_TYPE_HIVE = "hive";
+ public static final String ICEBERG_CATALOG_TYPE_REST = "rest";
+
+ public static final String HIVE_CONF_DIR = "hive-conf-dir";
+ public static final String HADOOP_CONF_DIR = "hadoop-conf-dir";
+ public static final String DEFAULT_DATABASE = "default-database";
+ public static final String DEFAULT_DATABASE_NAME = "default";
+ public static final String DEFAULT_CATALOG_NAME = "default_catalog";
+ public static final String BASE_NAMESPACE = "base-namespace";
+
+ /**
+ * Create an Iceberg {@link org.apache.iceberg.catalog.Catalog} loader to be used by this Flink
+ * catalog adapter.
+ *
+ * @param name Flink's catalog name
+ * @param properties Flink's catalog properties
+ * @param hadoopConf Hadoop configuration for catalog
+ * @return an Iceberg catalog loader
+ */
+ static CatalogLoader createCatalogLoader(
+ String name, Map properties, Configuration hadoopConf) {
+ String catalogImpl = properties.get(CatalogProperties.CATALOG_IMPL);
+ if (catalogImpl != null) {
+ String catalogType = properties.get(ICEBERG_CATALOG_TYPE);
+ Preconditions.checkArgument(
+ catalogType == null,
+ "Cannot create catalog %s, both catalog-type and catalog-impl are set: catalog-type=%s, catalog-impl=%s",
+ name,
+ catalogType,
+ catalogImpl);
+ return CatalogLoader.custom(name, properties, hadoopConf, catalogImpl);
+ }
+
+ String catalogType = properties.getOrDefault(ICEBERG_CATALOG_TYPE, ICEBERG_CATALOG_TYPE_HIVE);
+ switch (catalogType.toLowerCase(Locale.ROOT)) {
+ case ICEBERG_CATALOG_TYPE_HIVE:
+ // The values of properties 'uri', 'warehouse', 'hive-conf-dir' are allowed to be null, in
+ // that case it will
+ // fallback to parse those values from hadoop configuration which is loaded from classpath.
+ String hiveConfDir = properties.get(HIVE_CONF_DIR);
+ String hadoopConfDir = properties.get(HADOOP_CONF_DIR);
+ Configuration newHadoopConf = mergeHiveConf(hadoopConf, hiveConfDir, hadoopConfDir);
+ return CatalogLoader.hive(name, newHadoopConf, properties);
+
+ case ICEBERG_CATALOG_TYPE_HADOOP:
+ return CatalogLoader.hadoop(name, hadoopConf, properties);
+
+ case ICEBERG_CATALOG_TYPE_REST:
+ return CatalogLoader.rest(name, hadoopConf, properties);
+
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown catalog-type: " + catalogType + " (Must be 'hive', 'hadoop' or 'rest')");
+ }
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return FACTORY_IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return ImmutableSet.>builder().build();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return ImmutableSet.>builder().build();
+ }
+
+ @Override
+ public Catalog createCatalog(Context context) {
+ return createCatalog(context.getName(), context.getOptions(), clusterHadoopConf());
+ }
+
+ protected Catalog createCatalog(
+ String name, Map properties, Configuration hadoopConf) {
+ CatalogLoader catalogLoader = createCatalogLoader(name, properties, hadoopConf);
+ String defaultDatabase = properties.getOrDefault(DEFAULT_DATABASE, DEFAULT_DATABASE_NAME);
+
+ Namespace baseNamespace = Namespace.empty();
+ if (properties.containsKey(BASE_NAMESPACE)) {
+ baseNamespace = Namespace.of(properties.get(BASE_NAMESPACE).split("\\."));
+ }
+
+ boolean cacheEnabled =
+ PropertyUtil.propertyAsBoolean(
+ properties, CatalogProperties.CACHE_ENABLED, CatalogProperties.CACHE_ENABLED_DEFAULT);
+
+ long cacheExpirationIntervalMs =
+ PropertyUtil.propertyAsLong(
+ properties,
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS,
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS_OFF);
+ Preconditions.checkArgument(
+ cacheExpirationIntervalMs != 0,
+ "%s is not allowed to be 0.",
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS);
+
+ return new FlinkCatalog(
+ name,
+ defaultDatabase,
+ baseNamespace,
+ catalogLoader,
+ cacheEnabled,
+ cacheExpirationIntervalMs);
+ }
+
+ private static Configuration mergeHiveConf(
+ Configuration hadoopConf, String hiveConfDir, String hadoopConfDir) {
+ Configuration newConf = new Configuration(hadoopConf);
+ if (!Strings.isNullOrEmpty(hiveConfDir)) {
+ Preconditions.checkState(
+ Files.exists(Paths.get(hiveConfDir, "hive-site.xml")),
+ "There should be a hive-site.xml file under the directory %s",
+ hiveConfDir);
+ newConf.addResource(new Path(hiveConfDir, "hive-site.xml"));
+ } else {
+ // If don't provide the hive-site.xml path explicitly, it will try to load resource from
+ // classpath. If still
+ // couldn't load the configuration file, then it will throw exception in HiveCatalog.
+ URL configFile = CatalogLoader.class.getClassLoader().getResource("hive-site.xml");
+ if (configFile != null) {
+ newConf.addResource(configFile);
+ }
+ }
+
+ if (!Strings.isNullOrEmpty(hadoopConfDir)) {
+ Preconditions.checkState(
+ Files.exists(Paths.get(hadoopConfDir, "hdfs-site.xml")),
+ "Failed to load Hadoop configuration: missing %s",
+ Paths.get(hadoopConfDir, "hdfs-site.xml"));
+ newConf.addResource(new Path(hadoopConfDir, "hdfs-site.xml"));
+ Preconditions.checkState(
+ Files.exists(Paths.get(hadoopConfDir, "core-site.xml")),
+ "Failed to load Hadoop configuration: missing %s",
+ Paths.get(hadoopConfDir, "core-site.xml"));
+ newConf.addResource(new Path(hadoopConfDir, "core-site.xml"));
+ }
+
+ return newConf;
+ }
+
+ public static Configuration clusterHadoopConf() {
+ return HadoopUtils.getHadoopConfiguration(GlobalConfiguration.loadConfiguration());
+ }
+}
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfParser.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfParser.java
new file mode 100644
index 000000000000..7661372c88e8
--- /dev/null
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfParser.java
@@ -0,0 +1,297 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.util.TimeUtils;
+import org.apache.iceberg.Table;
+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;
+
+@Internal
+public class FlinkConfParser {
+
+ private final Map tableProperties;
+ private final Map options;
+ private final ReadableConfig readableConfig;
+
+ public FlinkConfParser(Table table, Map options, ReadableConfig readableConfig) {
+ this.tableProperties = table.properties();
+ this.options = options;
+ this.readableConfig = readableConfig;
+ }
+
+ public FlinkConfParser(Map options, ReadableConfig readableConfig) {
+ this.tableProperties = ImmutableMap.of();
+ this.options = options;
+ this.readableConfig = readableConfig;
+ }
+
+ public BooleanConfParser booleanConf() {
+ return new BooleanConfParser();
+ }
+
+ public IntConfParser intConf() {
+ return new IntConfParser();
+ }
+
+ public LongConfParser longConf() {
+ return new LongConfParser();
+ }
+
+ public DoubleConfParser doubleConf() {
+ return new DoubleConfParser();
+ }
+
+ public > EnumConfParser enumConfParser(Class enumClass) {
+ return new EnumConfParser<>(enumClass);
+ }
+
+ public StringConfParser stringConf() {
+ return new StringConfParser();
+ }
+
+ public DurationConfParser durationConf() {
+ return new DurationConfParser();
+ }
+
+ public class BooleanConfParser extends ConfParser {
+ private Boolean defaultValue;
+
+ @Override
+ protected BooleanConfParser self() {
+ return this;
+ }
+
+ public BooleanConfParser defaultValue(boolean value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public BooleanConfParser defaultValue(String value) {
+ this.defaultValue = Boolean.parseBoolean(value);
+ return self();
+ }
+
+ public boolean parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(Boolean::parseBoolean, defaultValue);
+ }
+ }
+
+ public class IntConfParser extends ConfParser {
+ private Integer defaultValue;
+
+ @Override
+ protected IntConfParser self() {
+ return this;
+ }
+
+ public IntConfParser defaultValue(int value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public int parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(Integer::parseInt, defaultValue);
+ }
+
+ public Integer parseOptional() {
+ return parse(Integer::parseInt, null);
+ }
+ }
+
+ public class LongConfParser extends ConfParser {
+ private Long defaultValue;
+
+ @Override
+ protected LongConfParser self() {
+ return this;
+ }
+
+ public LongConfParser defaultValue(long value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public long parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(Long::parseLong, defaultValue);
+ }
+
+ public Long parseOptional() {
+ return parse(Long::parseLong, null);
+ }
+ }
+
+ public class DoubleConfParser extends ConfParser {
+ private Double defaultValue;
+
+ @Override
+ protected DoubleConfParser self() {
+ return this;
+ }
+
+ public DoubleConfParser defaultValue(double value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public double parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(Double::parseDouble, defaultValue);
+ }
+
+ public Double parseOptional() {
+ return parse(Double::parseDouble, null);
+ }
+ }
+
+ public class StringConfParser extends ConfParser {
+ private String defaultValue;
+
+ @Override
+ protected StringConfParser self() {
+ return this;
+ }
+
+ public StringConfParser defaultValue(String value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public String parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(Function.identity(), defaultValue);
+ }
+
+ public String parseOptional() {
+ return parse(Function.identity(), null);
+ }
+ }
+
+ public class EnumConfParser> extends ConfParser, E> {
+ private E defaultValue;
+ private final Class enumClass;
+
+ EnumConfParser(Class enumClass) {
+ this.enumClass = enumClass;
+ }
+
+ @Override
+ protected EnumConfParser self() {
+ return this;
+ }
+
+ public EnumConfParser defaultValue(E value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public E parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(s -> Enum.valueOf(enumClass, s), defaultValue);
+ }
+
+ public E parseOptional() {
+ return parse(s -> Enum.valueOf(enumClass, s), null);
+ }
+ }
+
+ public class DurationConfParser extends ConfParser {
+ private Duration defaultValue;
+
+ @Override
+ protected DurationConfParser self() {
+ return this;
+ }
+
+ public DurationConfParser defaultValue(Duration value) {
+ this.defaultValue = value;
+ return self();
+ }
+
+ public Duration parse() {
+ Preconditions.checkArgument(defaultValue != null, "Default value cannot be null");
+ return parse(TimeUtils::parseDuration, defaultValue);
+ }
+
+ public Duration parseOptional() {
+ return parse(TimeUtils::parseDuration, null);
+ }
+ }
+
+ public abstract class ConfParser {
+ private final List optionNames = Lists.newArrayList();
+ private String tablePropertyName;
+ private ConfigOption configOption;
+
+ protected abstract ThisT self();
+
+ public ThisT option(String name) {
+ this.optionNames.add(name);
+ return self();
+ }
+
+ public ThisT flinkConfig(ConfigOption newConfigOption) {
+ this.configOption = newConfigOption;
+ return self();
+ }
+
+ public ThisT tableProperty(String name) {
+ this.tablePropertyName = name;
+ return self();
+ }
+
+ protected T parse(Function conversion, T defaultValue) {
+ if (!optionNames.isEmpty()) {
+ for (String optionName : optionNames) {
+ String optionValue = options.get(optionName);
+ if (optionValue != null) {
+ return conversion.apply(optionValue);
+ }
+ }
+ }
+
+ if (configOption != null) {
+ T propertyValue = readableConfig.get(configOption);
+ if (propertyValue != null) {
+ return propertyValue;
+ }
+ }
+
+ if (tablePropertyName != null) {
+ String propertyValue = tableProperties.get(tablePropertyName);
+ if (propertyValue != null) {
+ return conversion.apply(propertyValue);
+ }
+ }
+
+ return defaultValue;
+ }
+ }
+}
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
new file mode 100644
index 000000000000..97e2c70d348e
--- /dev/null
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.description.Description;
+import org.apache.flink.configuration.description.TextElement;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.iceberg.flink.source.assigner.SplitAssignerType;
+import org.apache.iceberg.util.ThreadPools;
+
+/**
+ * When constructing Flink Iceberg source via Java API, configs can be set in {@link Configuration}
+ * passed to source builder. E.g.
+ *
+ *
+ * configuration.setBoolean(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, true);
+ * FlinkSource.forRowData()
+ * .flinkConf(configuration)
+ * ...
+ *
+ *
+ * When using Flink SQL/table API, connector options can be set in Flink's {@link
+ * TableEnvironment}.
+ *
+ *
+ * TableEnvironment tEnv = createTableEnv();
+ * tEnv.getConfig()
+ * .getConfiguration()
+ * .setBoolean(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, true);
+ *
+ */
+public class FlinkConfigOptions {
+
+ private FlinkConfigOptions() {}
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "If is false, parallelism of source are set by config.\n"
+ + "If is true, source parallelism is inferred according to splits number.\n");
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM_MAX =
+ ConfigOptions.key("table.exec.iceberg.infer-source-parallelism.max")
+ .intType()
+ .defaultValue(100)
+ .withDescription("Sets max infer parallelism for source operator.");
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO =
+ ConfigOptions.key("table.exec.iceberg.expose-split-locality-info")
+ .booleanType()
+ .noDefaultValue()
+ .withDescription(
+ "Expose split host information to use Flink's locality aware split assigner.");
+
+ public static final ConfigOption SOURCE_READER_FETCH_BATCH_RECORD_COUNT =
+ ConfigOptions.key("table.exec.iceberg.fetch-batch-record-count")
+ .intType()
+ .defaultValue(2048)
+ .withDescription("The target number of records for Iceberg reader fetch batch.");
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_WORKER_POOL_SIZE =
+ ConfigOptions.key("table.exec.iceberg.worker-pool-size")
+ .intType()
+ .defaultValue(ThreadPools.WORKER_THREAD_POOL_SIZE)
+ .withDescription("The size of workers pool used to plan or scan manifests.");
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE =
+ ConfigOptions.key("table.exec.iceberg.use-flip27-source")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription("Use the FLIP-27 based Iceberg source implementation.");
+
+ public static final ConfigOption TABLE_EXEC_ICEBERG_USE_V2_SINK =
+ ConfigOptions.key("table.exec.iceberg.use-v2-sink")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Use the SinkV2 API based Iceberg sink implementation.");
+
+ public static final ConfigOption TABLE_EXEC_SPLIT_ASSIGNER_TYPE =
+ ConfigOptions.key("table.exec.iceberg.split-assigner-type")
+ .enumType(SplitAssignerType.class)
+ .defaultValue(SplitAssignerType.SIMPLE)
+ .withDescription(
+ Description.builder()
+ .text("Split assigner type that determine how splits are assigned to readers.")
+ .linebreak()
+ .list(
+ TextElement.text(
+ SplitAssignerType.SIMPLE
+ + ": simple assigner that doesn't provide any guarantee on order or locality."))
+ .build());
+}
diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
new file mode 100644
index 000000000000..a08cb19c8c4b
--- /dev/null
+++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.iceberg.util.JsonUtil;
+
+@Internal
+public class FlinkCreateTableOptions {
+ private final String catalogName;
+ private final String catalogDb;
+ private final String catalogTable;
+
+ private FlinkCreateTableOptions(String catalogName, String catalogDb, String catalogTable) {
+ this.catalogName = catalogName;
+ this.catalogDb = catalogDb;
+ this.catalogTable = catalogTable;
+ }
+
+ public static final ConfigOption CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption