You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by st...@apache.org on 2026/06/29 21:09:48 UTC
(iceberg) 02/02: Flink: Port white-listed Catalog properties fix to v1.19
This is an automated email from the ASF dual-hosted git repository.
stevenzwu pushed a commit to branch 1.10.x
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit d93cd81301fda8815c69c6c9887643074e7884d6
Author: Steven Wu <st...@gmail.com>
AuthorDate: Mon Jun 29 11:56:11 2026 -0700
Flink: Port white-listed Catalog properties fix to v1.19
Mirrors the v1.20/v2.0 backport in #16966 (which backports #16728) to
the v1.19 module, which exists only on the 1.10.x branch.
---
.../org/apache/iceberg/flink/FlinkCatalog.java | 10 +++-------
.../apache/iceberg/flink/FlinkCatalogFactory.java | 1 -
.../iceberg/flink/FlinkCreateTableOptions.java | 23 +++-------------------
.../iceberg/flink/FlinkDynamicTableFactory.java | 11 +++++------
.../org/apache/iceberg/flink/CatalogTestBase.java | 1 +
.../iceberg/flink/TestFlinkCatalogTable.java | 4 +++-
.../iceberg/flink/source/TestIcebergSourceSql.java | 15 +++++++++++---
7 files changed, 27 insertions(+), 38 deletions(-)
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
index 4c44961df1..42b9281e93 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -100,7 +100,6 @@ public class FlinkCatalog extends AbstractCatalog {
private final Namespace baseNamespace;
private final SupportsNamespaces asNamespaceCatalog;
private final Closeable closeable;
- private final Map<String, String> catalogProps;
private final boolean cacheEnabled;
public FlinkCatalog(
@@ -108,12 +107,10 @@ public class FlinkCatalog extends AbstractCatalog {
String defaultDatabase,
Namespace baseNamespace,
CatalogLoader catalogLoader,
- Map<String, String> catalogProps,
boolean cacheEnabled,
long cacheExpirationIntervalMs) {
super(catalogName, defaultDatabase);
this.catalogLoader = catalogLoader;
- this.catalogProps = catalogProps;
this.baseNamespace = baseNamespace;
this.cacheEnabled = cacheEnabled;
@@ -340,13 +337,12 @@ public class FlinkCatalog extends AbstractCatalog {
Table table = loadIcebergTable(tablePath);
// Flink's CREATE TABLE LIKE clause relies on properties sent back here to create new table.
- // Inorder to create such table in non iceberg catalog, we need to send across catalog
- // properties also.
// As Flink API accepts only Map<String, String> for props, here we are serializing catalog
- // props as json string to distinguish between catalog and table properties in createTable.
+ // 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(), catalogProps);
+ getName(), tablePath.getDatabaseName(), tablePath.getObjectName());
Map<String, String> tableProps = table.properties();
if (tableProps.containsKey(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY)
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
index dd065617bd..fe4008a13c 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
@@ -168,7 +168,6 @@ public class FlinkCatalogFactory implements CatalogFactory {
defaultDatabase,
baseNamespace,
catalogLoader,
- properties,
cacheEnabled,
cacheExpirationIntervalMs);
}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
index ab69ec5adc..77101d8459 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
@@ -18,7 +18,6 @@
*/
package org.apache.iceberg.flink;
-import java.util.Map;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.iceberg.util.JsonUtil;
@@ -27,14 +26,11 @@ class FlinkCreateTableOptions {
private final String catalogName;
private final String catalogDb;
private final String catalogTable;
- private final Map<String, String> catalogProps;
- private FlinkCreateTableOptions(
- String catalogName, String catalogDb, String catalogTable, Map<String, String> props) {
+ private FlinkCreateTableOptions(String catalogName, String catalogDb, String catalogTable) {
this.catalogName = catalogName;
this.catalogDb = catalogDb;
this.catalogTable = catalogTable;
- this.catalogProps = props;
}
public static final ConfigOption<String> CATALOG_NAME =
@@ -61,25 +57,17 @@ class FlinkCreateTableOptions {
.noDefaultValue()
.withDescription("Table name managed in the underlying iceberg catalog and database.");
- public static final ConfigOption<Map<String, String>> CATALOG_PROPS =
- ConfigOptions.key("catalog-props")
- .mapType()
- .noDefaultValue()
- .withDescription("Properties for the underlying catalog for iceberg table.");
-
public static final String SRC_CATALOG_PROPS_KEY = "src-catalog";
public static final String CONNECTOR_PROPS_KEY = "connector";
public static final String LOCATION_KEY = "location";
- static String toJson(
- String catalogName, String catalogDb, String catalogTable, Map<String, String> catalogProps) {
+ static String toJson(String catalogName, String catalogDb, String catalogTable) {
return JsonUtil.generate(
gen -> {
gen.writeStartObject();
gen.writeStringField(CATALOG_NAME.key(), catalogName);
gen.writeStringField(CATALOG_DATABASE.key(), catalogDb);
gen.writeStringField(CATALOG_TABLE.key(), catalogTable);
- JsonUtil.writeStringMap(CATALOG_PROPS.key(), catalogProps, gen);
gen.writeEndObject();
},
false);
@@ -92,9 +80,8 @@ class FlinkCreateTableOptions {
String catalogName = JsonUtil.getString(CATALOG_NAME.key(), node);
String catalogDb = JsonUtil.getString(CATALOG_DATABASE.key(), node);
String catalogTable = JsonUtil.getString(CATALOG_TABLE.key(), node);
- Map<String, String> catalogProps = JsonUtil.getStringMap(CATALOG_PROPS.key(), node);
- return new FlinkCreateTableOptions(catalogName, catalogDb, catalogTable, catalogProps);
+ return new FlinkCreateTableOptions(catalogName, catalogDb, catalogTable);
});
}
@@ -109,8 +96,4 @@ class FlinkCreateTableOptions {
String catalogTable() {
return catalogTable;
}
-
- Map<String, String> catalogProps() {
- return catalogProps;
- }
}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index 3b9cc0baff..225663d156 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -194,13 +194,13 @@ public class FlinkDynamicTableFactory
}
/**
- * Merges source catalog properties with connector properties. Iceberg Catalog properties are
- * serialized as json in FlinkCatalog#getTable to be able to isolate catalog props from iceberg
- * table props, Here, we flatten and merge them back to use to create catalog.
+ * Merges source catalog properties (catalog name, database, table) with connector properties.
+ * Source catalog name, database, table are serialized as json in FlinkCatalog#getTable to be able
+ * to isolate them from iceberg table props, Here, we flatten and merge them back.
*
* @param tableProps the existing table properties
- * @return a map of merged properties, with source catalog properties taking precedence when keys
- * conflict
+ * @return a map of merged properties, defaulting to source catalog name, database and table
+ * unless overridden.
*/
private static Map<String, String> mergeSrcCatalogProps(Map<String, String> tableProps) {
String srcCatalogProps = tableProps.get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY);
@@ -214,7 +214,6 @@ public class FlinkDynamicTableFactory
FlinkCreateTableOptions.CATALOG_DATABASE.key(), createTableOptions.catalogDb());
mergedProps.put(
FlinkCreateTableOptions.CATALOG_TABLE.key(), createTableOptions.catalogTable());
- mergedProps.putAll(createTableOptions.catalogProps());
tableProps.forEach(
(k, v) -> {
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
index 062ff68d5d..e45abe25f9 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
@@ -89,6 +89,7 @@ public abstract class CatalogTestBase extends TestBase {
config.put(CatalogProperties.URI, getURI(hiveConf));
}
config.put(CatalogProperties.WAREHOUSE_LOCATION, String.format("file://%s", warehouseRoot()));
+ config.put("extra-catalog-prop", "extra-value");
this.flinkDatabase = catalogName + "." + DATABASE;
this.icebergNamespace =
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index 2cfee87e46..2edc8fa60d 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -227,13 +227,15 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
.column("id", DataTypes.BIGINT())
.build());
- String srcCatalogProps = FlinkCreateTableOptions.toJson(catalogName, DATABASE, "tl", config);
+ String srcCatalogProps = FlinkCreateTableOptions.toJson(catalogName, DATABASE, "tl");
Map<String, String> options = catalogTable.getOptions();
assertThat(options)
.containsEntry(
FlinkCreateTableOptions.CONNECTOR_PROPS_KEY,
FlinkDynamicTableFactory.FACTORY_IDENTIFIER)
.containsEntry(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY, srcCatalogProps);
+ assertThat(options.get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY))
+ .doesNotContain("extra-catalog-prop", "extra-value");
}
@TestTemplate
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 0cdaf8371c..6cfc18868a 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
@@ -179,7 +179,10 @@ public class TestIcebergSourceSql extends TestSqlBase {
List<Record> expected = generateExpectedRecords(false);
SqlHelpers.sql(
getTableEnv(),
- "create table `default_catalog`.`default_database`.flink_table LIKE iceberg_catalog.`default`.%s",
+ "create table `default_catalog`.`default_database`.flink_table "
+ + "WITH ('catalog-type'='hadoop', 'warehouse'='%s') "
+ + "LIKE iceberg_catalog.`default`.%s",
+ CATALOG_EXTENSION.warehouse(),
TestFixtures.TABLE);
// Read from table in flink catalog
@@ -199,8 +202,11 @@ public class TestIcebergSourceSql extends TestSqlBase {
getStreamingTableEnv(),
"CREATE TABLE %s "
+ "(eventTS AS CAST(t1 AS TIMESTAMP(3)), "
- + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) LIKE iceberg_catalog.`default`.%s",
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) "
+ + "WITH ('catalog-type'='hadoop', 'warehouse'='%s') "
+ + "LIKE iceberg_catalog.`default`.%s",
flinkTable,
+ CATALOG_EXTENSION.warehouse(),
TestFixtures.TABLE);
assertThatThrownBy(() -> SqlHelpers.sql(getStreamingTableEnv(), "SELECT * FROM %s", flinkTable))
@@ -218,8 +224,11 @@ public class TestIcebergSourceSql extends TestSqlBase {
getStreamingTableEnv(),
"CREATE TABLE %s "
+ "(eventTS AS CAST(t1 AS TIMESTAMP(3)), "
- + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH ('watermark-column'='t1') LIKE iceberg_catalog.`default`.%s",
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) "
+ + "WITH ('catalog-type'='hadoop', 'warehouse'='%s', 'watermark-column'='t1') "
+ + "LIKE iceberg_catalog.`default`.%s",
flinkTable,
+ CATALOG_EXTENSION.warehouse(),
TestFixtures.TABLE);
TestHelpers.assertRecordsWithOrder(