You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/08/29 05:10:37 UTC
(iceberg) 01/02: Flink: Forward-port CatalogType test refactor to 2.3
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit e58b5b264e25a2ae877bc8d77fa2117b97de8fd4
Author: Talat Uyarer <ta...@apache.org>
AuthorDate: Fri Aug 28 12:12:42 2026 -0700
Flink: Forward-port CatalogType test refactor to 2.3
---
.../org/apache/iceberg/flink/CatalogTestBase.java | 32 ++++++++++++++++++----
.../flink/TestFlinkCatalogTablePartitions.java | 6 ++--
.../apache/iceberg/flink/TestFlinkTableSink.java | 16 +++++++----
.../flink/TestFlinkTableSinkCompaction.java | 13 ++++-----
.../org/apache/iceberg/flink/TestFlinkUpsert.java | 11 ++++----
.../apache/iceberg/flink/TestFlinkUuidType.java | 8 +++---
.../flink/TestFlinkVariantShreddingType.java | 6 ++--
.../apache/iceberg/flink/TestFlinkVariantType.java | 6 ++--
.../apache/iceberg/flink/TestIcebergConnector.java | 2 +-
.../flink/actions/TestRewriteDataFilesAction.java | 6 ++--
.../flink/source/TestFlinkMetaDataTable.java | 6 ++--
.../source/TestMetadataTableReadableMetrics.java | 6 ++--
12 files changed, 67 insertions(+), 51 deletions(-)
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
index e45abe25f9..bd34e62817 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/CatalogTestBase.java
@@ -43,12 +43,30 @@ import org.junit.jupiter.api.io.TempDir;
@ExtendWith(ParameterizedTestExtension.class)
public abstract class CatalogTestBase extends TestBase {
+ /** Catalog implementations that SQL test suites can run against. */
+ public enum CatalogType {
+ HIVE("testhive"),
+ HADOOP("testhadoop");
+
+ private final String catalogNamePrefix;
+
+ CatalogType(String catalogNamePrefix) {
+ this.catalogNamePrefix = catalogNamePrefix;
+ }
+
+ String catalogName(Namespace namespace) {
+ return namespace.isEmpty()
+ ? catalogNamePrefix
+ : catalogNamePrefix + "_" + Joiner.on('_').join(namespace.levels());
+ }
+ }
+
protected static final String DATABASE = "db";
@TempDir protected File hiveWarehouse;
@TempDir protected File hadoopWarehouse;
@Parameter(index = 0)
- protected String catalogName;
+ protected CatalogType catalogType;
@Parameter(index = 1)
protected Namespace baseNamespace;
@@ -57,21 +75,23 @@ public abstract class CatalogTestBase extends TestBase {
protected SupportsNamespaces validationNamespaceCatalog;
protected Map<String, String> config = Maps.newHashMap();
+ protected String catalogName;
protected String flinkDatabase;
protected Namespace icebergNamespace;
protected boolean isHadoopCatalog;
- @Parameters(name = "catalogName={0}, baseNamespace={1}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}")
protected static List<Object[]> parameters() {
return Arrays.asList(
- new Object[] {"testhive", Namespace.empty()},
- new Object[] {"testhadoop", Namespace.empty()},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1")});
+ new Object[] {CatalogType.HIVE, Namespace.empty()},
+ new Object[] {CatalogType.HADOOP, Namespace.empty()},
+ new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1")});
}
@BeforeEach
public void before() {
- this.isHadoopCatalog = catalogName.startsWith("testhadoop");
+ this.catalogName = catalogType.catalogName(baseNamespace);
+ this.isHadoopCatalog = catalogType == CatalogType.HADOOP;
this.validationCatalog =
isHadoopCatalog
? new HadoopCatalog(hiveConf, "file:" + hadoopWarehouse.getPath())
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java
index e69e1ac4d7..94e7348e4e 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java
@@ -47,16 +47,16 @@ public class TestFlinkCatalogTablePartitions extends CatalogTestBase {
@Parameter(index = 3)
private Boolean cacheEnabled;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}, cacheEnabled={3}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}, format={2}, cacheEnabled={3}")
protected static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
for (FileFormat format :
new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
for (Boolean cacheEnabled : new Boolean[] {true, false}) {
for (Object[] catalogParams : CatalogTestBase.parameters()) {
- String catalogName = (String) catalogParams[0];
+ CatalogType catalogType = (CatalogType) catalogParams[0];
Namespace baseNamespace = (Namespace) catalogParams[1];
- parameters.add(new Object[] {catalogName, baseNamespace, format, cacheEnabled});
+ parameters.add(new Object[] {catalogType, baseNamespace, format, cacheEnabled});
}
}
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
index 572ff68202..f15e222072 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
@@ -54,18 +54,18 @@ public class TestFlinkTableSink extends CatalogTestBase {
private boolean useV2Sink;
@Parameters(
- name = "catalogName={0}, baseNamespace={1}, format={2}, isStreaming={3}, useV2Sink={4}")
+ name = "catalogType={0}, baseNamespace={1}, format={2}, isStreaming={3}, useV2Sink={4}")
public static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
for (FileFormat format :
new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
for (Boolean isStreaming : new Boolean[] {true, false}) {
for (Object[] catalogParams : CatalogTestBase.parameters()) {
- String catalogName = (String) catalogParams[0];
+ CatalogType catalogType = (CatalogType) catalogParams[0];
Namespace baseNamespace = (Namespace) catalogParams[1];
parameters.add(
new Object[] {
- catalogName, baseNamespace, format, isStreaming, false /* don't use v2 sink */
+ catalogType, baseNamespace, format, isStreaming, false /* don't use v2 sink */
});
}
}
@@ -74,10 +74,14 @@ public class TestFlinkTableSink extends CatalogTestBase {
for (FileFormat format :
new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
for (Boolean isStreaming : new Boolean[] {true, false}) {
- String catalogName = "testhadoop_basenamespace";
- Namespace baseNamespace = Namespace.of("l0", "l1");
parameters.add(
- new Object[] {catalogName, baseNamespace, format, isStreaming, true /* use v2 sink */});
+ new Object[] {
+ CatalogType.HADOOP,
+ Namespace.of("l0", "l1"),
+ format,
+ isStreaming,
+ true /* use v2 sink */
+ });
}
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
index e5e478fe4a..afc9a4c2be 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
@@ -90,20 +90,17 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
@Parameter(index = 3)
private String lockType;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, userSqlHint={2}, lockType={3}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}, userSqlHint={2}, lockType={3}")
public static List<Object[]> parameters() {
return Arrays.asList(
new Object[] {
- "testhadoop_basenamespace", Namespace.of("l0", "l1"), true, LockConfig.JdbcLockConfig.JDBC
+ CatalogType.HADOOP, Namespace.of("l0", "l1"), true, LockConfig.JdbcLockConfig.JDBC
},
new Object[] {
- "testhadoop_basenamespace",
- Namespace.of("l0", "l1"),
- false,
- LockConfig.JdbcLockConfig.JDBC
+ CatalogType.HADOOP, Namespace.of("l0", "l1"), false, LockConfig.JdbcLockConfig.JDBC
},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), true, ""},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), false, ""});
+ new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1"), true, ""},
+ new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1"), false, ""});
}
@Override
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUpsert.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUpsert.java
index e6ca9c5c74..a83863d904 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUpsert.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUpsert.java
@@ -55,7 +55,7 @@ public class TestFlinkUpsert extends CatalogTestBase {
private TableEnvironment tEnv;
@Parameters(
- name = "catalogName={0}, baseNamespace={1}, format={2}, isStreaming={3}, formatVersion={4} ")
+ name = "catalogType={0}, baseNamespace={1}, format={2}, isStreaming={3}, formatVersion={4} ")
public static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
for (FileFormat format :
@@ -63,11 +63,10 @@ public class TestFlinkUpsert extends CatalogTestBase {
for (int version : org.apache.iceberg.TestHelpers.V2_AND_ABOVE) {
for (Boolean isStreaming : new Boolean[] {true, false}) {
// Only test with one catalog as this is a file operation concern.
- // FlinkCatalogTestBase requires the catalog name start with testhadoop if using hadoop
- // catalog.
- String catalogName = "testhadoop";
- Namespace baseNamespace = Namespace.of("default");
- parameters.add(new Object[] {catalogName, baseNamespace, format, isStreaming, version});
+ parameters.add(
+ new Object[] {
+ CatalogType.HADOOP, Namespace.of("default"), format, isStreaming, version
+ });
}
}
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUuidType.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUuidType.java
index cac258e7ad..ffaaccdad6 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUuidType.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkUuidType.java
@@ -72,12 +72,12 @@ class TestFlinkUuidType extends CatalogTestBase {
@Parameter(index = 2)
private FileFormat fileFormat;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, fileFormat={2}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}, fileFormat={2}")
protected static List<Object[]> parameters() {
return Arrays.asList(
- new Object[] {"testhadoop", Namespace.empty(), FileFormat.PARQUET},
- new Object[] {"testhadoop", Namespace.empty(), FileFormat.AVRO},
- new Object[] {"testhadoop", Namespace.empty(), FileFormat.ORC});
+ new Object[] {CatalogType.HADOOP, Namespace.empty(), FileFormat.PARQUET},
+ new Object[] {CatalogType.HADOOP, Namespace.empty(), FileFormat.AVRO},
+ new Object[] {CatalogType.HADOOP, Namespace.empty(), FileFormat.ORC});
}
@Override
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
index 6cb035ddff..b0a08a516a 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
@@ -57,11 +57,11 @@ class TestFlinkVariantShreddingType extends CatalogTestBase {
private static final String TABLE_NAME = "test_table";
private Table icebergTable;
- @Parameters(name = "catalogName={0}, baseNamespace={1}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}")
protected static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
- parameters.add(new Object[] {"testhadoop", Namespace.empty()});
- parameters.add(new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1")});
+ parameters.add(new Object[] {CatalogType.HADOOP, Namespace.empty()});
+ parameters.add(new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1")});
return parameters;
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantType.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantType.java
index decdb12b46..50e2cff2af 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantType.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantType.java
@@ -55,12 +55,12 @@ class TestFlinkVariantType extends CatalogTestBase {
private Table icebergTable;
@TempDir private Path warehouseDir;
- @Parameters(name = "catalogName={0}, baseNamespace={1}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}")
protected static List<Object[]> parameters() {
return Arrays.asList(
// For now hive metadata is not supported variant, so we only test hadoop catalog
- new Object[] {"testhadoop", Namespace.empty()},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1")});
+ new Object[] {CatalogType.HADOOP, Namespace.empty()},
+ new Object[] {CatalogType.HADOOP, Namespace.of("l0", "l1")});
}
@Override
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
index 92cc21880b..d11f2d3c27 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
@@ -328,7 +328,7 @@ public class TestIcebergConnector extends TestBase {
}
private boolean isHiveCatalog() {
- return "testhive".equalsIgnoreCase(catalogName);
+ return "hive".equals(properties.get("catalog-type"));
}
private boolean isDefaultDatabaseName() {
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
index 3d6ed921b1..a09efee96f 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
@@ -94,16 +94,16 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
return super.getTableEnv();
}
- @Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}, formatVersion={3}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}, format={2}, formatVersion={3}")
public static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
for (FileFormat format :
new FileFormat[] {FileFormat.AVRO, FileFormat.ORC, FileFormat.PARQUET}) {
for (Object[] catalogParams : CatalogTestBase.parameters()) {
for (int version : TestHelpers.V2_AND_ABOVE) {
- String catalogName = (String) catalogParams[0];
+ CatalogType catalogType = (CatalogType) catalogParams[0];
Namespace baseNamespace = (Namespace) catalogParams[1];
- parameters.add(new Object[] {catalogName, baseNamespace, format, version});
+ parameters.add(new Object[] {catalogType, baseNamespace, format, version});
}
}
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
index 6ae7bfb53a..186a2df8dc 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
@@ -81,14 +81,12 @@ public class TestFlinkMetaDataTable extends CatalogTestBase {
@Parameter(index = 2)
private Boolean isPartition;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, isPartition={2}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}, isPartition={2}")
protected static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
for (Boolean isPartition : new Boolean[] {true, false}) {
- String catalogName = "testhadoop";
- Namespace baseNamespace = Namespace.of("default");
- parameters.add(new Object[] {catalogName, baseNamespace, isPartition});
+ parameters.add(new Object[] {CatalogType.HADOOP, Namespace.of("default"), isPartition});
}
return parameters;
}
diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
index af3a79ff67..0ec01b20d3 100644
--- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
+++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
@@ -59,12 +59,10 @@ import org.junit.jupiter.api.io.TempDir;
public class TestMetadataTableReadableMetrics extends CatalogTestBase {
private static final String TABLE_NAME = "test_table";
- @Parameters(name = "catalogName={0}, baseNamespace={1}")
+ @Parameters(name = "catalogType={0}, baseNamespace={1}")
protected static List<Object[]> parameters() {
List<Object[]> parameters = Lists.newArrayList();
- String catalogName = "testhive";
- Namespace baseNamespace = Namespace.empty();
- parameters.add(new Object[] {catalogName, baseNamespace});
+ parameters.add(new Object[] {CatalogType.HIVE, Namespace.empty()});
return parameters;
}