This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new beb6901ae [hive] Support syncing Iceberg compatible metadata to Hive,
so Paimon table can be accessed from Iceberg Hive catalog (#4343)
beb6901ae is described below
commit beb6901ae8c776fe6c53d4ee98ec9b2f79001b59
Author: tsreaper <[email protected]>
AuthorDate: Tue Oct 22 00:01:21 2024 +0800
[hive] Support syncing Iceberg compatible metadata to Hive, so Paimon table
can be accessed from Iceberg Hive catalog (#4343)
---
docs/content/migration/iceberg-compatibility.md | 104 ++++++----
.../iceberg/AbstractIcebergCommitCallback.java | 38 +++-
.../paimon/iceberg/IcebergMetadataCommitter.java | 32 +++
.../iceberg/IcebergMetadataCommitterFactory.java | 28 +++
.../org/apache/paimon/iceberg/IcebergOptions.java | 24 ++-
.../iceberg/manifest/IcebergDataFileMeta.java | 2 +-
.../iceberg/manifest/IcebergManifestEntry.java | 2 +-
.../iceberg/manifest/IcebergManifestFileMeta.java | 2 +-
.../iceberg/manifest/IcebergPartitionSummary.java | 2 +-
.../paimon/iceberg/metadata/IcebergMetadata.java | 21 +-
.../flink/iceberg/FlinkIcebergITCaseBase.java | 89 ++++-----
.../iceberg/IcebergHiveMetadataCommitter.java | 216 +++++++++++++++++++++
.../IcebergHiveMetadataCommitterFactory.java | 35 ++++
.../services/org.apache.paimon.factories.Factory | 1 +
paimon-hive/paimon-hive-connector-2.3/pom.xml | 7 +
.../IcebergHive23MetadataCommitterITCase.java | 22 +++
paimon-hive/paimon-hive-connector-3.1/pom.xml | 7 +
.../IcebergHive31MetadataCommitterITCase.java | 22 +++
.../IcebergHiveMetadataCommitterITCaseBase.java | 162 ++++++++++++++++
paimon-hive/pom.xml | 1 +
20 files changed, 719 insertions(+), 98 deletions(-)
diff --git a/docs/content/migration/iceberg-compatibility.md
b/docs/content/migration/iceberg-compatibility.md
index 466f672a1..6a7103473 100644
--- a/docs/content/migration/iceberg-compatibility.md
+++ b/docs/content/migration/iceberg-compatibility.md
@@ -53,6 +53,7 @@ Set the following table options, so that Paimon tables can
generate Iceberg comp
<li>disabled: Disable Iceberg compatibility support.</li>
<li>table-location: Store Iceberg metadata in each table's
directory.</li>
<li>hadoop-catalog: Store Iceberg metadata in a separate directory.
This directory can be specified as the warehouse directory of an Iceberg Hadoop
catalog.</li>
+ <li>hive-catalog: Not only store Iceberg metadata like
hadoop-catalog, but also create Iceberg external table in Hive.</li>
</ul>
</td>
</tr>
@@ -131,18 +132,14 @@ Now let's query this Paimon table with Iceberg connector.
{{< tab "Flink SQL" >}}
```sql
-CREATE TABLE cities (
- country STRING,
- name STRING
-) WITH (
- 'connector' = 'iceberg',
+CREATE CATALOG iceberg_catalog WITH (
+ 'type' = 'iceberg',
'catalog-type' = 'hadoop',
- 'catalog-name' = 'test',
- 'catalog-database' = 'default',
- 'warehouse' = '<path-to-warehouse>/iceberg'
+ 'warehouse' = '<path-to-warehouse>/iceberg',
+ 'cache-enabled' = 'false' -- disable iceberg catalog caching to quickly
see the result
);
-SELECT * FROM cities WHERE country = 'germany';
+SELECT * FROM iceberg_catalog.`default`.cities WHERE country = 'germany';
/*
+----+--------------------------------+--------------------------------+
| op | country | name |
@@ -174,7 +171,7 @@ Let's insert more data and query again.
```sql
INSERT INTO paimon_catalog.`default`.cities VALUES ('usa', 'houston'),
('germany', 'munich');
-SELECT * FROM cities WHERE country = 'germany';
+SELECT * FROM iceberg_catalog.`default`.cities WHERE country = 'germany';
/*
+----+--------------------------------+--------------------------------+
| op | country | name |
@@ -225,19 +222,14 @@ CREATE TABLE paimon_catalog.`default`.orders (
INSERT INTO paimon_catalog.`default`.orders VALUES (1, 'SUBMITTED', CAST(NULL
AS DOUBLE)), (2, 'COMPLETED', 200.0), (3, 'SUBMITTED', CAST(NULL AS DOUBLE));
-CREATE TABLE orders (
- order_id BIGINT,
- status STRING,
- payment DOUBLE
-) WITH (
- 'connector' = 'iceberg',
+CREATE CATALOG iceberg_catalog WITH (
+ 'type' = 'iceberg',
'catalog-type' = 'hadoop',
- 'catalog-name' = 'test',
- 'catalog-database' = 'default',
- 'warehouse' = '<path-to-warehouse>/iceberg'
+ 'warehouse' = '<path-to-warehouse>/iceberg',
+ 'cache-enabled' = 'false' -- disable iceberg catalog caching to quickly
see the result
);
-SELECT * FROM orders WHERE status = 'COMPLETED';
+SELECT * FROM iceberg_catalog.`default`.orders WHERE status = 'COMPLETED';
/*
+----+----------------------+--------------------------------+--------------------------------+
| op | order_id | status |
payment |
@@ -248,7 +240,7 @@ SELECT * FROM orders WHERE status = 'COMPLETED';
INSERT INTO paimon_catalog.`default`.orders VALUES (1, 'COMPLETED', 100.0);
-SELECT * FROM orders WHERE status = 'COMPLETED';
+SELECT * FROM iceberg_catalog.`default`.orders WHERE status = 'COMPLETED';
/*
+----+----------------------+--------------------------------+--------------------------------+
| op | order_id | status |
payment |
@@ -345,25 +337,65 @@ You can configure the following table option, so that
Paimon is forced to perfor
Note that full compaction is a resource-consuming process, so the value of
this table option should not be too small.
We recommend full compaction to be performed once or twice per hour.
+## Access Paimon Table from Iceberg Hive Catalog
+
+When creating Paimon table, set `'metadata.iceberg.storage' = 'hive-catalog'`.
+This option value not only store Iceberg metadata like hadoop-catalog, but
also create Iceberg external table in Hive.
+This Paimon table can be accessed from Iceberg Hive catalog later.
+
+To provide information about Hive metastore,
+you also need to set some (or all) of the following table options when
creating Paimon table.
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Option</th>
+ <th class="text-left" style="width: 5%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 60%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>metadata.iceberg.uri</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>String</td>
+ <td>Hive metastore uri for Iceberg Hive catalog.</td>
+ </tr>
+ <tr>
+ <td><h5>metadata.iceberg.hive-conf-dir</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>String</td>
+ <td>hive-conf-dir for Iceberg Hive catalog.</td>
+ </tr>
+ <tr>
+ <td><h5>metadata.iceberg.hadoop-conf-dir</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>String</td>
+ <td>hadoop-conf-dir for Iceberg Hive catalog.</td>
+ </tr>
+ </tbody>
+</table>
+
## Supported Types
Paimon Iceberg compatibility currently supports the following data types.
-| Paimon Data Type | Iceberg Data Type |
-|----------------------------|-------------------|
-| `BOOLEAN` | `boolean` |
-| `INT` | `int` |
-| `BIGINT` | `long` |
-| `FLOAT` | `float` |
-| `DOUBLE` | `double` |
-| `DECIMAL` | `decimal` |
-| `CHAR` | `string` |
-| `VARCHAR` | `string` |
-| `BINARY` | `binary` |
-| `VARBINARY` | `binary` |
-| `DATE` | `date` |
-| `TIMESTAMP`* | `timestamp` |
-| `TIMESTAMP_LTZ`* | `timestamptz` |
+| Paimon Data Type | Iceberg Data Type |
+|-------------------|-------------------|
+| `BOOLEAN` | `boolean` |
+| `INT` | `int` |
+| `BIGINT` | `long` |
+| `FLOAT` | `float` |
+| `DOUBLE` | `double` |
+| `DECIMAL` | `decimal` |
+| `CHAR` | `string` |
+| `VARCHAR` | `string` |
+| `BINARY` | `binary` |
+| `VARBINARY` | `binary` |
+| `DATE` | `date` |
+| `TIMESTAMP`* | `timestamp` |
+| `TIMESTAMP_LTZ`* | `timestamptz` |
*: `TIMESTAMP` and `TIMESTAMP_LTZ` type only support precision from 4 to 6
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java
index 97eae5073..1b952c171 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/AbstractIcebergCommitCallback.java
@@ -23,6 +23,8 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.factories.FactoryException;
+import org.apache.paimon.factories.FactoryUtil;
import org.apache.paimon.fs.Path;
import org.apache.paimon.iceberg.manifest.IcebergConversions;
import org.apache.paimon.iceberg.manifest.IcebergDataFileMeta;
@@ -57,6 +59,8 @@ import org.apache.paimon.utils.ManifestReadThreadPool;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
@@ -84,9 +88,11 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
protected final FileStoreTable table;
private final String commitUser;
+
private final IcebergPathFactory pathFactory;
- private final FileStorePathFactory fileStorePathFactory;
+ private final @Nullable IcebergMetadataCommitter metadataCommitter;
+ private final FileStorePathFactory fileStorePathFactory;
private final IcebergManifestFile manifestFile;
private final IcebergManifestList manifestList;
@@ -105,6 +111,7 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
this.pathFactory = new IcebergPathFactory(new
Path(table.location(), "metadata"));
break;
case HADOOP_CATALOG:
+ case HIVE_CATALOG:
Path dbPath = table.location().getParent();
final String dbSuffix = ".db";
if (dbPath.getName().endsWith(dbSuffix)) {
@@ -127,6 +134,19 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
"Unknown storage type " + storageType.name());
}
+ IcebergMetadataCommitterFactory metadataCommitterFactory;
+ try {
+ metadataCommitterFactory =
+ FactoryUtil.discoverFactory(
+
AbstractIcebergCommitCallback.class.getClassLoader(),
+ IcebergMetadataCommitterFactory.class,
+ storageType.toString());
+ } catch (FactoryException ignore) {
+ metadataCommitterFactory = null;
+ }
+ this.metadataCommitter =
+ metadataCommitterFactory == null ? null :
metadataCommitterFactory.create(table);
+
this.fileStorePathFactory = table.store().pathFactory();
this.manifestFile = IcebergManifestFile.create(table, pathFactory);
this.manifestList = IcebergManifestList.create(table, pathFactory);
@@ -241,13 +261,19 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
IcebergPartitionField.FIRST_FIELD_ID -
1),
Collections.singletonList(snapshot),
(int) snapshotId);
-
table.fileIO().tryToWriteAtomic(pathFactory.toMetadataPath(snapshotId),
metadata.toJson());
+
+ Path metadataPath = pathFactory.toMetadataPath(snapshotId);
+ table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson());
table.fileIO()
.overwriteFileUtf8(
new Path(pathFactory.metadataDirectory(),
VERSION_HINT_FILENAME),
String.valueOf(snapshotId));
expireAllBefore(snapshotId);
+
+ if (metadataCommitter != null) {
+ metadataCommitter.commitMetadata(metadataPath, null);
+ }
}
private List<IcebergManifestEntry> dataSplitToManifestEntries(
@@ -384,7 +410,9 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
baseMetadata.lastPartitionId(),
snapshots,
(int) snapshotId);
-
table.fileIO().tryToWriteAtomic(pathFactory.toMetadataPath(snapshotId),
metadata.toJson());
+
+ Path metadataPath = pathFactory.toMetadataPath(snapshotId);
+ table.fileIO().tryToWriteAtomic(metadataPath, metadata.toJson());
table.fileIO()
.overwriteFileUtf8(
new Path(pathFactory.metadataDirectory(),
VERSION_HINT_FILENAME),
@@ -396,6 +424,10 @@ public abstract class AbstractIcebergCommitCallback
implements CommitCallback {
new
Path(toExpireExceptLast.get(i).manifestList()).getName(),
new Path(toExpireExceptLast.get(i +
1).manifestList()).getName());
}
+
+ if (metadataCommitter != null) {
+ metadataCommitter.commitMetadata(metadataPath, baseMetadataPath);
+ }
}
private interface FileChangesCollector {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitter.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitter.java
new file mode 100644
index 000000000..2fe131498
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitter.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.fs.Path;
+
+import javax.annotation.Nullable;
+
+/**
+ * Commit Iceberg metadata to metastore. Each kind of Iceberg catalog should
have its own
+ * implementation.
+ */
+public interface IcebergMetadataCommitter {
+
+ void commitMetadata(Path newMetadataPath, @Nullable Path baseMetadataPath);
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitterFactory.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitterFactory.java
new file mode 100644
index 000000000..dc168ff09
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergMetadataCommitterFactory.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.factories.Factory;
+import org.apache.paimon.table.FileStoreTable;
+
+/** Factory to create {@link IcebergMetadataCommitter}. */
+public interface IcebergMetadataCommitterFactory extends Factory {
+
+ IcebergMetadataCommitter create(FileStoreTable table);
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
index cbc78be76..769ce6b16 100644
--- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergOptions.java
@@ -52,6 +52,24 @@ public class IcebergOptions {
"If number of small Iceberg metadata files exceeds
this limit, "
+ "always trigger metadata compaction
regardless of their total size.");
+ public static final ConfigOption<String> URI =
+ key("metadata.iceberg.uri")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Hive metastore uri for Iceberg Hive
catalog.");
+
+ public static final ConfigOption<String> HIVE_CONF_DIR =
+ key("metadata.iceberg.hive-conf-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("hive-conf-dir for Iceberg Hive
catalog.");
+
+ public static final ConfigOption<String> HADOOP_CONF_DIR =
+ key("metadata.iceberg.hadoop-conf-dir")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("hadoop-conf-dir for Iceberg Hive
catalog.");
+
/** Where to store Iceberg metadata. */
public enum StorageType implements DescribedEnum {
DISABLED("disabled", "Disable Iceberg compatibility support."),
@@ -59,7 +77,11 @@ public class IcebergOptions {
HADOOP_CATALOG(
"hadoop-catalog",
"Store Iceberg metadata in a separate directory. "
- + "This directory can be specified as the warehouse
directory of an Iceberg Hadoop catalog.");
+ + "This directory can be specified as the warehouse
directory of an Iceberg Hadoop catalog."),
+ HIVE_CATALOG(
+ "hive-catalog",
+ "Not only store Iceberg metadata like hadoop-catalog, "
+ + "but also create Iceberg external table in Hive.");
private final String value;
private final String description;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
index c9a71668e..10dbf3d23 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMeta.java
@@ -205,7 +205,7 @@ public class IcebergDataFileMeta {
128,
"upper_bounds",
DataTypes.MAP(DataTypes.INT().notNull(),
DataTypes.BYTES().notNull())));
- return new RowType(fields);
+ return new RowType(false, fields);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
index 213ec0131..1e8108a94 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestEntry.java
@@ -118,7 +118,7 @@ public class IcebergManifestEntry {
fields.add(new DataField(4, "file_sequence_number",
DataTypes.BIGINT()));
fields.add(
new DataField(2, "data_file",
IcebergDataFileMeta.schema(partitionType).notNull()));
- return new RowType(fields);
+ return new RowType(false, fields);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
index 01b3b96ac..fd9c2daf7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
@@ -183,7 +183,7 @@ public class IcebergManifestFileMeta {
fields.add(
new DataField(
508, "partitions",
DataTypes.ARRAY(IcebergPartitionSummary.schema())));
- return new RowType(fields);
+ return new RowType(false, fields);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergPartitionSummary.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergPartitionSummary.java
index e47a8fe00..9d2731860 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergPartitionSummary.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergPartitionSummary.java
@@ -69,7 +69,7 @@ public class IcebergPartitionSummary {
fields.add(new DataField(518, "contains_nan", DataTypes.BOOLEAN()));
fields.add(new DataField(510, "lower_bound", DataTypes.BYTES()));
fields.add(new DataField(511, "upper_bound", DataTypes.BYTES()));
- return (RowType) new RowType(fields).notNull();
+ return new RowType(false, fields);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergMetadata.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergMetadata.java
index a3af25a1c..86fb4a5df 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergMetadata.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/metadata/IcebergMetadata.java
@@ -27,9 +27,13 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGet
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
/**
@@ -57,6 +61,7 @@ public class IcebergMetadata {
private static final String FIELD_DEFAULT_SORT_ORDER_ID =
"default-sort-order-id";
private static final String FIELD_SNAPSHOTS = "snapshots";
private static final String FIELD_CURRENT_SNAPSHOT_ID =
"current-snapshot-id";
+ private static final String FIELD_PROPERTIES = "properties";
@JsonProperty(FIELD_FORMAT_VERSION)
private final int formatVersion;
@@ -103,6 +108,10 @@ public class IcebergMetadata {
@JsonProperty(FIELD_CURRENT_SNAPSHOT_ID)
private final int currentSnapshotId;
+ @JsonProperty(FIELD_PROPERTIES)
+ @Nullable
+ private final Map<String, String> properties;
+
public IcebergMetadata(
String tableUuid,
String location,
@@ -129,7 +138,8 @@ public class IcebergMetadata {
Collections.singletonList(new IcebergSortOrder()),
IcebergSortOrder.ORDER_ID,
snapshots,
- currentSnapshotId);
+ currentSnapshotId,
+ new HashMap<>());
}
@JsonCreator
@@ -148,7 +158,8 @@ public class IcebergMetadata {
@JsonProperty(FIELD_SORT_ORDERS) List<IcebergSortOrder> sortOrders,
@JsonProperty(FIELD_DEFAULT_SORT_ORDER_ID) int defaultSortOrderId,
@JsonProperty(FIELD_SNAPSHOTS) List<IcebergSnapshot> snapshots,
- @JsonProperty(FIELD_CURRENT_SNAPSHOT_ID) int currentSnapshotId) {
+ @JsonProperty(FIELD_CURRENT_SNAPSHOT_ID) int currentSnapshotId,
+ @JsonProperty(FIELD_PROPERTIES) @Nullable Map<String, String>
properties) {
this.formatVersion = formatVersion;
this.tableUuid = tableUuid;
this.location = location;
@@ -164,6 +175,7 @@ public class IcebergMetadata {
this.defaultSortOrderId = defaultSortOrderId;
this.snapshots = snapshots;
this.currentSnapshotId = currentSnapshotId;
+ this.properties = properties;
}
@JsonGetter(FIELD_FORMAT_VERSION)
@@ -241,6 +253,11 @@ public class IcebergMetadata {
return currentSnapshotId;
}
+ @JsonGetter(FIELD_PROPERTIES)
+ public Map<String, String> properties() {
+ return properties == null ? new HashMap<>() : properties;
+ }
+
public IcebergSnapshot currentSnapshot() {
for (IcebergSnapshot snapshot : snapshots) {
if (snapshot.snapshotId() == currentSnapshotId) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
index d76394c9e..413a404c4 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/iceberg/FlinkIcebergITCaseBase.java
@@ -78,21 +78,18 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.await();
tEnv.executeSql(
- "CREATE TABLE T (\n"
- + " pt INT,\n"
- + " k INT,\n"
- + " v1 INT,\n"
- + " v2 STRING\n"
- + ") PARTITIONED BY (pt) WITH (\n"
- + " 'connector' = 'iceberg',\n"
+ "CREATE CATALOG iceberg WITH (\n"
+ + " 'type' = 'iceberg',\n"
+ " 'catalog-type' = 'hadoop',\n"
- + " 'catalog-name' = 'test',\n"
- + " 'catalog-database' = 'default',\n"
+ " 'warehouse' = '"
+ warehouse
- + "/iceberg'\n"
+ + "/iceberg',\n"
+ + " 'cache-enabled' = 'false'\n"
+ ")");
- assertThat(collect(tEnv.executeSql("SELECT v1, k, v2, pt FROM T ORDER
BY pt, k")))
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT v1, k, v2, pt FROM
iceberg.`default`.T ORDER BY pt, k")))
.containsExactly(
Row.of(100, 10, "apple", 1),
Row.of(110, 11, "banana", 1),
@@ -106,7 +103,10 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ "(2, 20, 201, 'blue'), "
+ "(2, 22, 221, 'yellow')")
.await();
- assertThat(collect(tEnv.executeSql("SELECT v1, k, v2, pt FROM T ORDER
BY pt, k")))
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT v1, k, v2, pt FROM
iceberg.`default`.T ORDER BY pt, k")))
.containsExactly(
Row.of(101, 10, "red", 1),
Row.of(110, 11, "banana", 1),
@@ -147,19 +147,15 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.await();
tEnv.executeSql(
- "CREATE TABLE cities (\n"
- + " country STRING,\n"
- + " name STRING\n"
- + ") WITH (\n"
- + " 'connector' = 'iceberg',\n"
+ "CREATE CATALOG iceberg WITH (\n"
+ + " 'type' = 'iceberg',\n"
+ " 'catalog-type' = 'hadoop',\n"
- + " 'catalog-name' = 'test',\n"
- + " 'catalog-database' = 'default',\n"
+ " 'warehouse' = '"
+ warehouse
- + "/iceberg'\n"
+ + "/iceberg',\n"
+ + " 'cache-enabled' = 'false'\n"
+ ")");
- assertThat(collect(tEnv.executeSql("SELECT name, country FROM
cities")))
+ assertThat(collect(tEnv.executeSql("SELECT name, country FROM
iceberg.`default`.cities")))
.containsExactlyInAnyOrder(
Row.of("new york", "usa"),
Row.of("chicago", "usa"),
@@ -171,7 +167,10 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ "('usa', 'houston'), "
+ "('germany', 'munich')")
.await();
- assertThat(collect(tEnv.executeSql("SELECT name FROM cities WHERE
country = 'germany'")))
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT name FROM
iceberg.`default`.cities WHERE country = 'germany'")))
.containsExactlyInAnyOrder(Row.of("berlin"),
Row.of("hamburg"), Row.of("munich"));
}
@@ -217,28 +216,15 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.await();
tEnv.executeSql(
- "CREATE TABLE T (\n"
- + " pt INT,\n"
- + " id INT,"
- + " v_int INT,\n"
- + " v_boolean BOOLEAN,\n"
- + " v_bigint BIGINT,\n"
- + " v_float FLOAT,\n"
- + " v_double DOUBLE,\n"
- + " v_decimal DECIMAL(8, 3),\n"
- + " v_varchar STRING,\n"
- + " v_varbinary VARBINARY(20),\n"
- + " v_date DATE,\n"
- + " v_timestamp TIMESTAMP(6)\n"
- + ") PARTITIONED BY (pt) WITH (\n"
- + " 'connector' = 'iceberg',\n"
+ "CREATE CATALOG iceberg WITH (\n"
+ + " 'type' = 'iceberg',\n"
+ " 'catalog-type' = 'hadoop',\n"
- + " 'catalog-name' = 'test',\n"
- + " 'catalog-database' = 'default',\n"
+ " 'warehouse' = '"
+ warehouse
- + "/iceberg'\n"
+ + "/iceberg',\n"
+ + " 'cache-enabled' = 'false'\n"
+ ")");
+ tEnv.executeSql("USE CATALOG iceberg");
assertThat(collect(tEnv.executeSql("SELECT id FROM T where pt = 1")))
.containsExactly(Row.of(1));
assertThat(collect(tEnv.executeSql("SELECT id FROM T where v_int =
1")))
@@ -373,19 +359,15 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
.await();
tEnv.executeSql(
- "CREATE TABLE cities (\n"
- + " country STRING,\n"
- + " name STRING\n"
- + ") WITH (\n"
- + " 'connector' = 'iceberg',\n"
+ "CREATE CATALOG iceberg WITH (\n"
+ + " 'type' = 'iceberg',\n"
+ " 'catalog-type' = 'hadoop',\n"
- + " 'catalog-name' = 'test',\n"
- + " 'catalog-database' = 'default',\n"
+ " 'warehouse' = '"
+ warehouse
- + "/iceberg'\n"
+ + "/iceberg',\n"
+ + " 'cache-enabled' = 'false'\n"
+ ")");
- assertThat(collect(tEnv.executeSql("SELECT name, country FROM
cities")))
+ assertThat(collect(tEnv.executeSql("SELECT name, country FROM
iceberg.`default`.cities")))
.containsExactlyInAnyOrder(Row.of("new york", "usa"),
Row.of("berlin", "germany"));
tEnv.executeSql(
@@ -393,7 +375,7 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ "('usa', 'chicago'), "
+ "('germany', 'hamburg')")
.await();
- assertThat(collect(tEnv.executeSql("SELECT name, country FROM
cities")))
+ assertThat(collect(tEnv.executeSql("SELECT name, country FROM
iceberg.`default`.cities")))
.containsExactlyInAnyOrder(
Row.of("new york", "usa"),
Row.of("chicago", "usa"),
@@ -407,7 +389,7 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ "('usa', 'houston'), "
+ "('germany', 'munich')")
.await();
- assertThat(collect(tEnv.executeSql("SELECT name, country FROM
cities")))
+ assertThat(collect(tEnv.executeSql("SELECT name, country FROM
iceberg.`default`.cities")))
.containsExactlyInAnyOrder(Row.of("houston", "usa"),
Row.of("munich", "germany"));
tEnv.executeSql(
@@ -415,7 +397,10 @@ public abstract class FlinkIcebergITCaseBase extends
AbstractTestBase {
+ "('usa', 'san francisco'), "
+ "('germany', 'cologne')")
.await();
- assertThat(collect(tEnv.executeSql("SELECT name FROM cities WHERE
country = 'germany'")))
+ assertThat(
+ collect(
+ tEnv.executeSql(
+ "SELECT name FROM
iceberg.`default`.cities WHERE country = 'germany'")))
.containsExactlyInAnyOrder(Row.of("munich"),
Row.of("cologne"));
}
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitter.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitter.java
new file mode 100644
index 000000000..d913f729e
--- /dev/null
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitter.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.client.ClientPool;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.hive.HiveCatalog;
+import org.apache.paimon.hive.HiveCatalogFactory;
+import org.apache.paimon.hive.HiveTypeUtils;
+import org.apache.paimon.hive.pool.CachedClientPool;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.stream.Collectors;
+
+/**
+ * {@link IcebergMetadataCommitter} to commit Iceberg metadata to Hive
metastore, so the table can
+ * be visited by Iceberg's Hive catalog.
+ */
+public class IcebergHiveMetadataCommitter implements IcebergMetadataCommitter {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergHiveMetadataCommitter.class);
+
+ private final FileStoreTable table;
+ private final Identifier identifier;
+ private final ClientPool<IMetaStoreClient, TException> clients;
+
+ public IcebergHiveMetadataCommitter(FileStoreTable table) {
+ this.table = table;
+ this.identifier =
+ Preconditions.checkNotNull(
+ table.catalogEnvironment().identifier(),
+ "If you want to sync Paimon Iceberg compatible
metadata to Hive, "
+ + "you must use a Paimon table created from a
Paimon catalog, "
+ + "instead of a temporary table.");
+ Preconditions.checkArgument(
+ identifier.getBranchName() == null,
+ "Paimon Iceberg compatibility currently does not support
branches.");
+
+ Options options = new Options(table.options());
+ String uri = options.get(IcebergOptions.URI);
+ String hiveConfDir = options.get(IcebergOptions.HIVE_CONF_DIR);
+ String hadoopConfDir = options.get(IcebergOptions.HADOOP_CONF_DIR);
+ Configuration hadoopConf = new Configuration();
+
hadoopConf.setClassLoader(IcebergHiveMetadataCommitter.class.getClassLoader());
+ HiveConf hiveConf = HiveCatalog.createHiveConf(hiveConfDir,
hadoopConfDir, hadoopConf);
+
+ table.options().forEach(hiveConf::set);
+ if (uri != null) {
+ hiveConf.set(HiveConf.ConfVars.METASTOREURIS.varname, uri);
+ }
+
+ if (hiveConf.get(HiveConf.ConfVars.METASTOREURIS.varname) == null) {
+ LOG.error(
+ "Can't find hive metastore uri to connect: "
+ + "either set {} for paimon table or set
hive.metastore.uris "
+ + "in hive-site.xml or hadoop configurations. "
+ + "Will use empty metastore uris, which means we
may use a embedded metastore. "
+ + "This may cause unpredictable consensus
problem.",
+ IcebergOptions.URI.key());
+ }
+
+ this.clients =
+ new CachedClientPool(
+ hiveConf,
+ options,
+
HiveCatalogFactory.METASTORE_CLIENT_CLASS.defaultValue());
+ }
+
+ @Override
+ public void commitMetadata(Path newMetadataPath, @Nullable Path
baseMetadataPath) {
+ try {
+ commitMetadataImpl(newMetadataPath, baseMetadataPath);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void commitMetadataImpl(Path newMetadataPath, @Nullable Path
baseMetadataPath)
+ throws Exception {
+ if (!databaseExists(identifier.getDatabaseName())) {
+ createDatabase(identifier.getDatabaseName());
+ }
+
+ Table hiveTable;
+ if (tableExists(identifier)) {
+ hiveTable =
+ clients.run(
+ client ->
+ client.getTable(
+ identifier.getDatabaseName(),
+ identifier.getTableName()));
+ } else {
+ hiveTable = createTable(newMetadataPath);
+ }
+
+ hiveTable.getParameters().put("metadata_location",
newMetadataPath.toString());
+ if (baseMetadataPath != null) {
+ hiveTable
+ .getParameters()
+ .put("previous_metadata_location",
baseMetadataPath.toString());
+ }
+
+ clients.execute(
+ client ->
+ client.alter_table(
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ hiveTable,
+ true));
+ }
+
+ private boolean databaseExists(String databaseName) throws Exception {
+ try {
+ clients.run(client -> client.getDatabase(databaseName));
+ return true;
+ } catch (NoSuchObjectException ignore) {
+ return false;
+ }
+ }
+
+ private void createDatabase(String databaseName) throws Exception {
+ Database database = new Database();
+ database.setName(databaseName);
+ clients.execute(client -> client.createDatabase(database));
+ }
+
+ private boolean tableExists(Identifier identifier) throws Exception {
+ return clients.run(
+ client ->
+ client.tableExists(
+ identifier.getDatabaseName(),
identifier.getTableName()));
+ }
+
+ private Table createTable(Path metadataPath) throws Exception {
+ long currentTimeMillis = System.currentTimeMillis();
+ Table hiveTable =
+ new Table(
+ identifier.getTableName(),
+ identifier.getDatabaseName(),
+ // current linux user
+ System.getProperty("user.name"),
+ (int) (currentTimeMillis / 1000),
+ (int) (currentTimeMillis / 1000),
+ Integer.MAX_VALUE,
+ new StorageDescriptor(),
+ Collections.emptyList(),
+ new HashMap<>(),
+ null,
+ null,
+ "EXTERNAL_TABLE");
+
+ hiveTable.getParameters().put("DO_NOT_UPDATE_STATS", "true");
+ hiveTable.getParameters().put("EXTERNAL", "TRUE");
+ hiveTable.getParameters().put("table_type", "ICEBERG");
+
+ StorageDescriptor sd = hiveTable.getSd();
+ sd.setLocation(metadataPath.getParent().getParent().toString());
+ sd.setCols(
+ table.schema().fields().stream()
+ .map(this::convertToFieldSchema)
+ .collect(Collectors.toList()));
+ sd.setInputFormat("org.apache.hadoop.mapred.FileInputFormat");
+ sd.setOutputFormat("org.apache.hadoop.mapred.FileOutputFormat");
+
+ SerDeInfo serDeInfo = new SerDeInfo();
+
serDeInfo.setSerializationLib("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe");
+ hiveTable.getSd().setSerdeInfo(serDeInfo);
+
+ clients.execute(client -> client.createTable(hiveTable));
+ return hiveTable;
+ }
+
+ private FieldSchema convertToFieldSchema(DataField dataField) {
+ return new FieldSchema(
+ dataField.name(),
+ HiveTypeUtils.toTypeInfo(dataField.type()).getTypeName(),
+ dataField.description());
+ }
+}
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterFactory.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterFactory.java
new file mode 100644
index 000000000..2ae279d3d
--- /dev/null
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterFactory.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.table.FileStoreTable;
+
+/** Factory to create {@link IcebergHiveMetadataCommitter}. */
+public class IcebergHiveMetadataCommitterFactory implements
IcebergMetadataCommitterFactory {
+
+ @Override
+ public String identifier() {
+ return IcebergOptions.StorageType.HIVE_CATALOG.toString();
+ }
+
+ @Override
+ public IcebergMetadataCommitter create(FileStoreTable table) {
+ return new IcebergHiveMetadataCommitter(table);
+ }
+}
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index baab92184..26f0944d9 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -15,3 +15,4 @@
org.apache.paimon.hive.HiveCatalogFactory
org.apache.paimon.hive.HiveCatalogLockFactory
+org.apache.paimon.iceberg.IcebergHiveMetadataCommitterFactory
diff --git a/paimon-hive/paimon-hive-connector-2.3/pom.xml
b/paimon-hive/paimon-hive-connector-2.3/pom.xml
index 22f93ed0e..a0f509b53 100644
--- a/paimon-hive/paimon-hive-connector-2.3/pom.xml
+++ b/paimon-hive/paimon-hive-connector-2.3/pom.xml
@@ -562,6 +562,13 @@ under the License.
</exclusions>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-flink-${iceberg.flink.version}</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/paimon-hive/paimon-hive-connector-2.3/src/test/java/org/apache/paimon/iceberg/IcebergHive23MetadataCommitterITCase.java
b/paimon-hive/paimon-hive-connector-2.3/src/test/java/org/apache/paimon/iceberg/IcebergHive23MetadataCommitterITCase.java
new file mode 100644
index 000000000..a9e4ba945
--- /dev/null
+++
b/paimon-hive/paimon-hive-connector-2.3/src/test/java/org/apache/paimon/iceberg/IcebergHive23MetadataCommitterITCase.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+/** IT cases for {@link IcebergHiveMetadataCommitter} in Hive 2.3. */
+public class IcebergHive23MetadataCommitterITCase extends
IcebergHiveMetadataCommitterITCaseBase {}
diff --git a/paimon-hive/paimon-hive-connector-3.1/pom.xml
b/paimon-hive/paimon-hive-connector-3.1/pom.xml
index 70afb0e4e..5383af90c 100644
--- a/paimon-hive/paimon-hive-connector-3.1/pom.xml
+++ b/paimon-hive/paimon-hive-connector-3.1/pom.xml
@@ -592,6 +592,13 @@ under the License.
</exclusions>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-flink-${iceberg.flink.version}</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/paimon-hive/paimon-hive-connector-3.1/src/test/java/org/apache/paimon/iceberg/IcebergHive31MetadataCommitterITCase.java
b/paimon-hive/paimon-hive-connector-3.1/src/test/java/org/apache/paimon/iceberg/IcebergHive31MetadataCommitterITCase.java
new file mode 100644
index 000000000..6f4b0afd1
--- /dev/null
+++
b/paimon-hive/paimon-hive-connector-3.1/src/test/java/org/apache/paimon/iceberg/IcebergHive31MetadataCommitterITCase.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+/** IT cases for {@link IcebergHiveMetadataCommitter} in Hive 3.1. */
+public class IcebergHive31MetadataCommitterITCase extends
IcebergHiveMetadataCommitterITCaseBase {}
diff --git
a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterITCaseBase.java
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterITCaseBase.java
new file mode 100644
index 000000000..fab227757
--- /dev/null
+++
b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/iceberg/IcebergHiveMetadataCommitterITCaseBase.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg;
+
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/** IT cases for {@link IcebergHiveMetadataCommitter}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public abstract class IcebergHiveMetadataCommitterITCaseBase {
+
+ @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+ @HiveSQL(files = {})
+ protected static HiveShell hiveShell;
+
+ private String path;
+
+ @Before
+ public void before() throws Exception {
+ path = folder.newFolder().toURI().toString();
+ }
+
+ @After
+ public void after() {
+ hiveShell.execute("DROP DATABASE IF EXISTS test_db CASCADE");
+ }
+
+ @Test
+ public void testPrimaryKeyTable() throws Exception {
+ TableEnvironment tEnv =
+ TableEnvironmentImpl.create(
+
EnvironmentSettings.newInstance().inBatchMode().build());
+ tEnv.executeSql(
+ "CREATE CATALOG my_paimon WITH ( 'type' = 'paimon',
'warehouse' = '"
+ + path
+ + "' )");
+ tEnv.executeSql("CREATE DATABASE my_paimon.test_db");
+ tEnv.executeSql(
+ "CREATE TABLE my_paimon.test_db.t ( pt INT, id INT, data
STRING, PRIMARY KEY (pt, id) NOT ENFORCED ) "
+ + "PARTITIONED BY (pt) WITH "
+ + "( 'metadata.iceberg.storage' = 'hive-catalog',
'metadata.iceberg.uri' = '', 'file.format' = 'avro', "
+ // make sure all changes are visible in iceberg
metadata
+ + " 'full-compaction.delta-commits' = '1' )");
+ tEnv.executeSql(
+ "INSERT INTO my_paimon.test_db.t VALUES "
+ + "(1, 1, 'apple'), (1, 2, 'pear'), (2, 1,
'cat'), (2, 2, 'dog')")
+ .await();
+
+ tEnv.executeSql(
+ "CREATE CATALOG my_iceberg WITH "
+ + "( 'type' = 'iceberg', 'catalog-type' = 'hive',
'uri' = '', 'warehouse' = '"
+ + path
+ + "', 'cache-enabled' = 'false' )");
+ Assert.assertEquals(
+ Arrays.asList(Row.of("pear", 2, 1), Row.of("dog", 2, 2)),
+ collect(
+ tEnv.executeSql(
+ "SELECT data, id, pt FROM my_iceberg.test_db.t
WHERE id = 2 ORDER BY pt, id")));
+
+ tEnv.executeSql(
+ "INSERT INTO my_paimon.test_db.t VALUES "
+ + "(1, 1, 'cherry'), (2, 2, 'elephant')")
+ .await();
+ Assert.assertEquals(
+ Arrays.asList(
+ Row.of(1, 1, "cherry"),
+ Row.of(1, 2, "pear"),
+ Row.of(2, 1, "cat"),
+ Row.of(2, 2, "elephant")),
+ collect(tEnv.executeSql("SELECT * FROM my_iceberg.test_db.t
ORDER BY pt, id")));
+ }
+
+ @Test
+ public void testAppendOnlyTable() throws Exception {
+ TableEnvironment tEnv =
+ TableEnvironmentImpl.create(
+
EnvironmentSettings.newInstance().inBatchMode().build());
+ tEnv.executeSql(
+ "CREATE CATALOG my_paimon WITH ( 'type' = 'paimon',
'warehouse' = '"
+ + path
+ + "' )");
+ tEnv.executeSql("CREATE DATABASE my_paimon.test_db");
+ tEnv.executeSql(
+ "CREATE TABLE my_paimon.test_db.t ( pt INT, id INT, data
STRING ) PARTITIONED BY (pt) WITH "
+ + "( 'metadata.iceberg.storage' = 'hive-catalog',
'metadata.iceberg.uri' = '', 'file.format' = 'avro' )");
+ tEnv.executeSql(
+ "INSERT INTO my_paimon.test_db.t VALUES "
+ + "(1, 1, 'apple'), (1, 2, 'pear'), (2, 1,
'cat'), (2, 2, 'dog')")
+ .await();
+
+ tEnv.executeSql(
+ "CREATE CATALOG my_iceberg WITH "
+ + "( 'type' = 'iceberg', 'catalog-type' = 'hive',
'uri' = '', 'warehouse' = '"
+ + path
+ + "', 'cache-enabled' = 'false' )");
+ Assert.assertEquals(
+ Arrays.asList(Row.of("pear", 2, 1), Row.of("dog", 2, 2)),
+ collect(
+ tEnv.executeSql(
+ "SELECT data, id, pt FROM my_iceberg.test_db.t
WHERE id = 2 ORDER BY pt, id")));
+
+ tEnv.executeSql(
+ "INSERT INTO my_paimon.test_db.t VALUES "
+ + "(1, 3, 'cherry'), (2, 3, 'elephant')")
+ .await();
+ Assert.assertEquals(
+ Arrays.asList(
+ Row.of("pear", 2, 1),
+ Row.of("cherry", 3, 1),
+ Row.of("dog", 2, 2),
+ Row.of("elephant", 3, 2)),
+ collect(
+ tEnv.executeSql(
+ "SELECT data, id, pt FROM my_iceberg.test_db.t
WHERE id > 1 ORDER BY pt, id")));
+ }
+
+ private List<Row> collect(TableResult result) throws Exception {
+ List<Row> rows = new ArrayList<>();
+ try (CloseableIterator<Row> it = result.collect()) {
+ while (it.hasNext()) {
+ rows.add(it.next());
+ }
+ }
+ return rows;
+ }
+}
diff --git a/paimon-hive/pom.xml b/paimon-hive/pom.xml
index 5c306f474..7d1d0f2c4 100644
--- a/paimon-hive/pom.xml
+++ b/paimon-hive/pom.xml
@@ -49,6 +49,7 @@ under the License.
<hiverunner.version>4.0.0</hiverunner.version>
<reflections.version>0.9.8</reflections.version>
<aws.version>1.12.319</aws.version>
+ <iceberg.flink.version>1.19</iceberg.flink.version>
</properties>
<dependencies>
