You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by ji...@apache.org on 2024/05/28 08:50:09 UTC
(amoro) branch master updated: [AMORO-2811][Bug][FLINK]: flink unified catalog doesn't support mixed_hive (#2814)
This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new fc5e9949d [AMORO-2811][Bug][FLINK]: flink unified catalog doesn't support mixed_hive (#2814)
fc5e9949d is described below
commit fc5e9949d67d85e0ffad51b8c334bea4d385fbc7
Author: Wang Tao <hz...@corp.netease.com>
AuthorDate: Tue May 28 16:50:05 2024 +0800
[AMORO-2811][Bug][FLINK]: flink unified catalog doesn't support mixed_hive (#2814)
* fix: flink unified catalog support multi formats
* fix: flink unified catalog support multi formats
* throw exception when we can't decide the table format
* fix table-format check logic
---
.../org/apache/amoro/flink/FlinkSchemaUtil.java | 2 +-
.../amoro/flink/catalog/FlinkUnifiedCatalog.java | 28 +++++++++++++++++++++-
.../amoro/hive/catalog/MixedHiveCatalog.java | 4 +++-
3 files changed, 31 insertions(+), 3 deletions(-)
diff --git a/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java b/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java
index 6091ebd7f..36513012e 100644
--- a/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java
+++ b/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java
@@ -33,7 +33,6 @@ import org.apache.amoro.table.PrimaryKeySpec;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.curator.shaded.com.google.common.collect.Maps;
import org.apache.flink.table.api.TableColumn;
import org.apache.flink.table.api.TableColumn.ComputedColumn;
import org.apache.flink.table.api.TableSchema;
@@ -45,6 +44,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.utils.LogicalTypeParser;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Types;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java b/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
index 8e31379da..21cafc46f 100644
--- a/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
+++ b/amoro-mixed-format/amoro-mixed-format-flink/amoro-mixed-format-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
@@ -28,6 +28,7 @@ import org.apache.amoro.NoSuchTableException;
import org.apache.amoro.TableFormat;
import org.apache.amoro.UnifiedCatalog;
import org.apache.amoro.client.AmsThriftUrl;
+import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
import org.apache.amoro.flink.catalog.factories.FlinkUnifiedCatalogFactory;
import org.apache.amoro.flink.catalog.factories.iceberg.IcebergFlinkCatalogFactory;
import org.apache.amoro.flink.catalog.factories.mixed.MixedCatalogFactory;
@@ -59,7 +60,10 @@ import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.Factory;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
@@ -68,6 +72,8 @@ import java.util.Optional;
/** This is a Flink catalog wrap a unified catalog. */
public class FlinkUnifiedCatalog extends AbstractCatalog {
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkUnifiedCatalog.class);
+
private final UnifiedCatalog unifiedCatalog;
private final String amsUri;
private final String amoroCatalogName;
@@ -221,8 +227,27 @@ public class FlinkUnifiedCatalog extends AbstractCatalog {
TableIdentifier tableIdentifier =
TableIdentifier.of(
unifiedCatalog.name(), tablePath.getDatabaseName(), tablePath.getObjectName());
+ String errorMessage =
+ String.format(
+ "Can't decide table format of table %s, Please specify 'table.format' "
+ + "in table properties",
+ tableIdentifier);
+
+ Preconditions.checkNotNull(format, errorMessage);
+ try {
+ unifiedCatalog.loadTable(tableIdentifier.getDatabase(), tableIdentifier.getTableName());
+ if (!ignoreIfExists) {
+ throw new TableAlreadyExistException(getName(), tablePath);
+ }
+ return;
+ } catch (NoSuchTableException e) {
+ // do nothing
+ }
+
+ final TableFormat catalogFormat = format;
AbstractCatalog catalog =
- getOriginalCatalog(format).orElseGet(() -> createOriginalCatalog(tableIdentifier, format));
+ getOriginalCatalog(format)
+ .orElseGet(() -> createOriginalCatalog(tableIdentifier, catalogFormat));
catalog.createTable(tablePath, table, ignoreIfExists);
}
@@ -462,6 +487,7 @@ public class FlinkUnifiedCatalog extends AbstractCatalog {
AbstractCatalog originalCatalog;
try {
+ context.getOptions().put(CatalogFactoryOptions.FLINK_TABLE_FORMATS.key(), tableFormat.name());
originalCatalog = (AbstractCatalog) catalogFactory.createCatalog(context);
} catch (CatalogException e) {
if (e.getMessage().contains("must implement createCatalog(Context)")) {
diff --git a/amoro-mixed-format/amoro-mixed-format-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveCatalog.java b/amoro-mixed-format/amoro-mixed-format-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveCatalog.java
index aea6e3502..10eab7f63 100644
--- a/amoro-mixed-format/amoro-mixed-format-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveCatalog.java
+++ b/amoro-mixed-format/amoro-mixed-format-hive/src/main/java/org/apache/amoro/hive/catalog/MixedHiveCatalog.java
@@ -578,7 +578,8 @@ public class MixedHiveCatalog implements MixedFormatCatalog {
// do some check for whether the table has been upgraded!!!
if (CompatibleHivePropertyUtil.propertyAsBoolean(
hiveTable.getParameters(), HiveTableProperties.MIXED_TABLE_FLAG, false)) {
- throw new IllegalArgumentException(
+ // flink will ignore the AlreadyExistsdException to continue
+ throw new org.apache.iceberg.exceptions.AlreadyExistsException(
String.format("Table %s has already been upgraded !", identifier));
}
}
@@ -586,6 +587,7 @@ public class MixedHiveCatalog implements MixedFormatCatalog {
LOG.info("No need to check hive table exist");
} else {
if (hiveTable != null) {
+ // we should throw the exception for stop create.
throw new IllegalArgumentException(
"Table is already existed in hive meta store:" + identifier);
}