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:47 UTC
(iceberg) 01/02: Flink: Backport Pass only white-listed Catalog properties to 1.20,2.0 (#16966)
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 d663f6f9dd110d52887712750488c079cb035972
Author: Swapna Marru <sw...@gmail.com>
AuthorDate: Thu Jun 25 22:36:29 2026 -0700
Flink: Backport Pass only white-listed Catalog properties to 1.20,2.0 (#16966)
Backports #16728
(cherry picked from commit 66150ef442be414c9580fc1c73962f1f9719d441)
---
.../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 +++++++++++---
.../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 | 11 +++--------
.../iceberg/flink/source/TestIcebergSourceSql.java | 15 +++++++++++---
14 files changed, 54 insertions(+), 83 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
index 4bb235b811..677d1aa871 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -98,7 +98,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(
@@ -106,12 +105,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;
@@ -338,13 +335,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.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
index dd065617bd..fe4008a13c 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
+++ b/flink/v1.20/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.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
index ab69ec5adc..77101d8459 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
+++ b/flink/v1.20/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.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index 3b9cc0baff..225663d156 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ b/flink/v1.20/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.20/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
index 062ff68d5d..e45abe25f9 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
+++ b/flink/v1.20/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.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index 2cfee87e46..2edc8fa60d 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v1.20/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.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 0cdaf8371c..6cfc18868a 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++ b/flink/v1.20/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(
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
index 4bb235b811..677d1aa871 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -98,7 +98,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(
@@ -106,12 +105,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;
@@ -338,13 +335,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/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
index 33cbc92dde..c1889dc6bd 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java
@@ -170,7 +170,6 @@ public class FlinkCatalogFactory implements CatalogFactory {
defaultDatabase,
baseNamespace,
catalogLoader,
- properties,
cacheEnabled,
cacheExpirationIntervalMs);
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
index ab69ec5adc..77101d8459 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java
+++ b/flink/v2.0/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/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index bd79c11560..1127337cee 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -193,13 +193,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);
@@ -213,7 +213,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/v2.0/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
index 062ff68d5d..e45abe25f9 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
+++ b/flink/v2.0/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/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index f7848a5d22..a292dd9b20 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -34,7 +34,6 @@ import org.apache.flink.table.api.Schema.UnresolvedPrimaryKey;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CommonCatalogOptions;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.iceberg.BaseTable;
@@ -228,19 +227,15 @@ public class TestFlinkCatalogTable extends CatalogTestBase {
.column("id", DataTypes.BIGINT())
.build());
- // `type` option is filtered out by Flink
- // https://github.com/apache/flink/blob/edc3d68736de73665440f4313ddcfd9142d8d42b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java#L378
- Map<String, String> filteredOptions = Maps.newHashMap(config);
- filteredOptions.remove(CommonCatalogOptions.CATALOG_TYPE.key());
-
- String srcCatalogProps =
- FlinkCreateTableOptions.toJson(catalogName, DATABASE, "tl", filteredOptions);
+ 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/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 0cdaf8371c..6cfc18868a 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++ b/flink/v2.0/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(