stevenzwu commented on code in PR #16728:
URL: https://github.com/apache/iceberg/pull/16728#discussion_r3438672030
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -218,9 +218,21 @@ public void testCreateTableLikeInDiffIcebergCatalog()
throws TableNotExistExcept
@TestTemplate
public void testCreateTableLikeInFlinkCatalog() throws
TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT)");
-
- sql("CREATE TABLE `default_catalog`.`default_database`.tl2 LIKE tl");
+ // Create a separate source catalog with an arbitrary extra property
+ // (catalog-cred), alongside the warehouse and uri set by the fixture, so
we can assert
+ // the LIKE allowlist drops every catalog property outside catalog-type,
catalog-impl.
+ String credentialKey = "catalog-cred";
+ String credentialValue = "secret-value";
+ String sourceCatalog = catalogName + "_with_cred";
+ String sourceDatabase = "db_with_cred";
+ String sourceFlinkDatabase = sourceCatalog + "." + sourceDatabase;
+ Map<String, String> sourceConfig = Maps.newHashMap(config);
+ sourceConfig.put(credentialKey, credentialValue);
+ sql("CREATE CATALOG %s WITH %s", sourceCatalog,
toWithClause(sourceConfig));
+ sql("CREATE DATABASE %s", sourceFlinkDatabase);
+ sql("CREATE TABLE %s.tl(id BIGINT)", sourceFlinkDatabase);
+
+ sql("CREATE TABLE `default_catalog`.`default_database`.tl2 LIKE %s.tl",
sourceFlinkDatabase);
Review Comment:
The previous version of this test exercised the simplest in-catalog LIKE —
`CREATE TABLE tl(id BIGINT)` followed by `CREATE TABLE
default_catalog.default_database.tl2 LIKE tl`. After this rewrite every run
goes through a fresh source catalog with an extra property, and the trivial
in-catalog LIKE path is no longer covered. Worth splitting: keep the original
minimal test, and use this method only for the allowlist filter assertions.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -231,17 +243,39 @@ public void testCreateTableLikeInFlinkCatalog() throws
TableNotExistException {
// `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);
+ // Only catalog-type / catalog-impl are retained for the LIKE target to
reconstruct the
+ // catalog; everything else (e.g. warehouse) must be supplied explicitly
via WITH.
+ Map<String, String> filteredOptions = Maps.newHashMap();
+ if (sourceConfig.containsKey(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)) {
+ filteredOptions.put(
+ FlinkCatalogFactory.ICEBERG_CATALOG_TYPE,
+ sourceConfig.get(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE));
+ }
+ if (sourceConfig.containsKey(CatalogProperties.CATALOG_IMPL)) {
+ filteredOptions.put(
+ CatalogProperties.CATALOG_IMPL,
sourceConfig.get(CatalogProperties.CATALOG_IMPL));
+ }
+
+ String expectedSourceCatalogProps =
+ FlinkCreateTableOptions.toJson(sourceCatalog, sourceDatabase, "tl",
filteredOptions);
Map<String, String> options = catalogTable.getOptions();
assertThat(options)
.containsEntry(
FlinkCreateTableOptions.CONNECTOR_PROPS_KEY,
FlinkDynamicTableFactory.FACTORY_IDENTIFIER)
- .containsEntry(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY,
srcCatalogProps);
+ .containsEntry(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY,
expectedSourceCatalogProps);
+
+ Map<String, String> srcCatalogProps =
+ FlinkCreateTableOptions.fromJson(
+ options.get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY));
+ assertThat(srcCatalogProps)
+ .doesNotContainKey(credentialKey)
+ .doesNotContainKey(CatalogProperties.URI)
+ .doesNotContainKey(CatalogProperties.WAREHOUSE_LOCATION);
+
+ sql("DROP TABLE IF EXISTS %s.tl", sourceFlinkDatabase);
+ dropDatabase(sourceFlinkDatabase, true);
+ dropCatalog(sourceCatalog, true);
Review Comment:
If any assertion above fails, this cleanup never runs and the `_with_cred`
catalog (plus its database) leaks into subsequent test methods. Either wrap the
body in `try { ... } finally { dropCatalog / dropDatabase / DROP TABLE }`, or
move the cleanup into an `@AfterEach` keyed on the names.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java:
##########
@@ -97,38 +83,30 @@ static String toJson(
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);
+
+ String catalogType = catalogProps.get(CATALOG_TYPE.key());
+ if (catalogType != null) {
+ gen.writeStringField(CATALOG_TYPE.key(), catalogType);
+ }
+
+ String catalogImpl =
catalogProps.get(CatalogProperties.CATALOG_IMPL);
+ if (catalogImpl != null) {
+ gen.writeStringField(CatalogProperties.CATALOG_IMPL, catalogImpl);
+ }
+
gen.writeEndObject();
},
false);
}
- static FlinkCreateTableOptions fromJson(String createTableOptions) {
+ static Map<String, String> fromJson(String createTableOptions) {
return JsonUtil.parse(
createTableOptions,
node -> {
- 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);
+ ImmutableMap.Builder<String, String> properties =
ImmutableMap.builder();
+ node.fieldNames()
+ .forEachRemaining(field -> properties.put(field,
JsonUtil.getString(field, node)));
Review Comment:
`JsonUtil.getString` throws when the field value is not a string. The
original format wrote a nested `catalog-props` JSON object, so any Flink
CatalogTable persisted from an earlier release will hit
`getString("catalog-props", node)` here on the next read and fail. Either skip
non-textual fields, or call out in the PR body that this is a one-way format
change and existing LIKE-derived tables need to be recreated.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java:
##########
@@ -97,38 +83,30 @@ static String toJson(
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);
+
+ String catalogType = catalogProps.get(CATALOG_TYPE.key());
Review Comment:
looks like we are still carrying over two catalog props. I thought we would
want to carry none for consistency.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -218,9 +218,21 @@ public void testCreateTableLikeInDiffIcebergCatalog()
throws TableNotExistExcept
@TestTemplate
public void testCreateTableLikeInFlinkCatalog() throws
TableNotExistException {
- sql("CREATE TABLE tl(id BIGINT)");
-
- sql("CREATE TABLE `default_catalog`.`default_database`.tl2 LIKE tl");
+ // Create a separate source catalog with an arbitrary extra property
Review Comment:
Two nits on this block:
- `source catalog with` has a double space.
- `credentialKey = "catalog-cred"` reads like a real Iceberg credential
property (e.g. `s3.access-key-id`). A generic name like `extraKey =
"extra-prop"` would more clearly signal "any non-allowlisted property" without
implying credential semantics.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java:
##########
@@ -94,6 +94,7 @@
*/
@Internal
public class FlinkCatalog extends AbstractCatalog {
+
Review Comment:
nit: avoid unnecessary white space change
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java:
##########
@@ -231,17 +243,39 @@ public void testCreateTableLikeInFlinkCatalog() throws
TableNotExistException {
// `type` option is filtered out by Flink
Review Comment:
The leading comment about Flink filtering `type` from destination options
originally paired with `filteredOptions.remove(CATALOG_TYPE.key())` — the test
used to remove `type` because Flink filters it. The new code below explicitly
*adds* `catalog-type` into `filteredOptions`, so this comment now precedes
logic that does the opposite, which is confusing on a read. Either drop the
comment, or rephrase so the link with Flink's `type` filtering reads naturally
next to the new logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
