You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/10/16 02:36:40 UTC
(paimon) branch master updated: [core] Support add partition for paimon (#4323)
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 cc3ed7f48 [core] Support add partition for paimon (#4323)
cc3ed7f48 is described below
commit cc3ed7f4859c9aa9d1af8c2cc8299089bb102f45
Author: xuzifu666 <12...@qq.com>
AuthorDate: Wed Oct 16 10:36:34 2024 +0800
[core] Support add partition for paimon (#4323)
---
.../org/apache/paimon/catalog/AbstractCatalog.java | 28 ++++++++++++++++++
.../org/apache/paimon/catalog/DelegateCatalog.java | 10 +++++++
.../java/org/apache/paimon/flink/FlinkCatalog.java | 19 +++++++++++--
.../java/org/apache/paimon/hive/HiveCatalog.java | 17 +++++++++++
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 20 +++++++++++++
.../paimon/spark/PaimonPartitionManagement.scala | 33 ++++++++++++++++++++--
.../spark/sql/DDLWithHiveCatalogTestBase.scala | 32 +++++++++++++++++++++
7 files changed, 155 insertions(+), 4 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 5ea714530..2542d3777 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -24,6 +24,7 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.lineage.LineageMetaFactory;
+import org.apache.paimon.metastore.MetastoreClient;
import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.operation.Lock;
import org.apache.paimon.options.Options;
@@ -40,12 +41,16 @@ import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.utils.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -67,6 +72,9 @@ public abstract class AbstractCatalog implements Catalog {
protected final FileIO fileIO;
protected final Map<String, String> tableDefaultOptions;
protected final Options catalogOptions;
+ public MetastoreClient metastoreClient;
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractCatalog.class);
@Nullable protected final LineageMetaFactory lineageMetaFactory;
@@ -154,6 +162,26 @@ public abstract class AbstractCatalog implements Catalog {
protected abstract Map<String, String> loadDatabasePropertiesImpl(String name)
throws DatabaseNotExistException;
+ public void createPartition(Identifier identifier, Map<String, String> partitionSpec)
+ throws TableNotExistException {
+ TableSchema tableSchema = getDataTableSchema(identifier);
+ if (!tableSchema.partitionKeys().isEmpty()
+ && new CoreOptions(tableSchema.options()).partitionedTableInMetastore()) {
+ try {
+ // Do not close client, it is for HiveCatalog
+ if (metastoreClient == null) {
+ throw new UnsupportedOperationException(
+ "Only Support HiveCatalog in create partition!");
+ }
+ metastoreClient.addPartition(new LinkedHashMap<>(partitionSpec));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ throw new RuntimeException("the table is not partitioned table in metastore!");
+ }
+ }
+
@Override
public void dropPartition(Identifier identifier, Map<String, String> partitionSpec)
throws TableNotExistException {
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index c2e36dea3..53d764f89 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -140,6 +140,16 @@ public class DelegateCatalog implements Catalog {
return wrapped.getTableLocation(identifier);
}
+ public void createPartition(Identifier identifier, Map<String, String> partitions)
+ throws TableNotExistException {
+ if (wrapped instanceof AbstractCatalog) {
+ ((AbstractCatalog) wrapped).createPartition(identifier, partitions);
+ } else {
+ throw new UnsupportedOperationException(
+ "Only Support HiveCatalog in create partition!");
+ }
+ }
+
@Override
public void dropPartition(Identifier identifier, Map<String, String> partitions)
throws TableNotExistException, PartitionNotExistException {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index 00b9b5931..7e006185c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -82,6 +82,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
@@ -100,6 +101,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.lang.reflect.Method;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -1145,8 +1147,21 @@ public class FlinkCatalog extends AbstractCatalog {
CatalogPartitionSpec partitionSpec,
CatalogPartition partition,
boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException();
+ throws CatalogException, PartitionAlreadyExistsException {
+ if (partitionExists(tablePath, partitionSpec)) {
+ if (!ignoreIfExists) {
+ throw new PartitionAlreadyExistsException(getName(), tablePath, partitionSpec);
+ }
+ }
+
+ try {
+ Identifier identifier = toIdentifier(tablePath);
+ Method func =
+ catalog.getClass().getMethod("createPartition", Identifier.class, Map.class);
+ func.invoke(catalog, identifier, partitionSpec.getPartitionSpec());
+ } catch (Exception e) {
+ throw new CatalogException(e);
+ }
}
@Override
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 4d9c482a2..e84222bd9 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -310,6 +310,23 @@ public class HiveCatalog extends AbstractCatalog {
return properties;
}
+ @Override
+ public void createPartition(Identifier identifier, Map<String, String> partitionSpec)
+ throws TableNotExistException {
+ try {
+ TableSchema tableSchema = getDataTableSchema(identifier);
+ metastoreClient =
+ new HiveMetastoreClient(
+ new Identifier(identifier.getDatabaseName(), identifier.getTableName()),
+ tableSchema,
+ clients);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ super.createPartition(identifier, partitionSpec);
+ }
+
@Override
public void dropPartition(Identifier identifier, Map<String, String> partitionSpec)
throws TableNotExistException {
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
index aa3062ec7..246822600 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
@@ -1348,6 +1348,26 @@ public abstract class HiveCatalogITCaseBase {
"ptb=2a/pta=2", "ptb=2b/pta=2", "ptb=3a/pta=3", "ptb=3b/pta=3");
}
+ @Test
+ public void testCreatePartitionsToMetastore() throws Exception {
+ prepareTestAddPartitionsToMetastore();
+
+ // add partition
+ tEnv.executeSql(
+ "ALTER TABLE t ADD PARTITION (ptb = '1c', pta = 1) PARTITION (ptb = '1d', pta = 6)")
+ .await();
+ assertThat(hiveShell.executeQuery("show partitions t"))
+ .containsExactlyInAnyOrder(
+ "ptb=1a/pta=1",
+ "ptb=1b/pta=1",
+ "ptb=1c/pta=1",
+ "ptb=1d/pta=6",
+ "ptb=2a/pta=2",
+ "ptb=2b/pta=2",
+ "ptb=3a/pta=3",
+ "ptb=3b/pta=3");
+ }
+
@Test
public void testAddPartitionsForTag() throws Exception {
tEnv.executeSql(
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
index 113ad4c3b..8bf816417 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
@@ -18,6 +18,8 @@
package org.apache.paimon.spark
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.metastore.MetastoreClient
import org.apache.paimon.operation.FileStoreCommit
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.sink.BatchWriteBuilder
@@ -30,7 +32,7 @@ import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.SupportsAtomicPartitionManagement
import org.apache.spark.sql.types.StructType
-import java.util.{Map => JMap, Objects, UUID}
+import java.util.{LinkedHashMap, Map => JMap, Objects, UUID}
import scala.collection.JavaConverters._
@@ -114,6 +116,33 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement {
override def createPartitions(
internalRows: Array[InternalRow],
maps: Array[JMap[String, String]]): Unit = {
- throw new UnsupportedOperationException("Create partition is not supported")
+ table match {
+ case fileStoreTable: FileStoreTable =>
+ val rowConverter = CatalystTypeConverters
+ .createToScalaConverter(CharVarcharUtils.replaceCharVarcharWithString(partitionSchema))
+ val rowDataPartitionComputer = new InternalRowPartitionComputer(
+ fileStoreTable.coreOptions().partitionDefaultName(),
+ partitionRowType,
+ table.partitionKeys().asScala.toArray)
+ val partitions = internalRows.map {
+ r =>
+ rowDataPartitionComputer
+ .generatePartValues(new SparkRow(partitionRowType, rowConverter(r).asInstanceOf[Row]))
+ .asInstanceOf[JMap[String, String]]
+ }
+ val metastoreClient: MetastoreClient =
+ fileStoreTable.catalogEnvironment().metastoreClientFactory().create
+ partitions.foreach {
+ partition =>
+ metastoreClient.addPartition(partition.asInstanceOf[LinkedHashMap[String, String]])
+ }
+ case _ =>
+ throw new UnsupportedOperationException("Only FileStoreTable supports create partitions.")
+ }
+ }
+
+ def getIdentifierFromTableName(tableName: String): Identifier = {
+ val name: Array[String] = tableName.split("\\.")
+ new Identifier(name.apply(0), name.apply(1))
}
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
index a371f3e31..0f2ef09bf 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
@@ -59,6 +59,38 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
}
}
+ test("Paimon DDL with hive catalog: create partition for paimon table sparkCatalogName") {
+ Seq(paimonHiveCatalogName).foreach {
+ catalogName =>
+ spark.sql(s"USE $catalogName")
+ withTempDir {
+ dBLocation =>
+ withDatabase("paimon_db") {
+ val comment = "this is a test comment"
+ spark.sql(
+ s"CREATE DATABASE paimon_db LOCATION '${dBLocation.getCanonicalPath}' COMMENT '$comment'")
+ Assertions.assertEquals(getDatabaseLocation("paimon_db"), dBLocation.getCanonicalPath)
+ Assertions.assertEquals(getDatabaseComment("paimon_db"), comment)
+
+ withTable("paimon_db.paimon_tbl") {
+ spark.sql(s"""
+ |CREATE TABLE paimon_db.paimon_tbl (id STRING, name STRING, pt STRING)
+ |USING PAIMON
+ |PARTITIONED BY (name, pt)
+ |TBLPROPERTIES('metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+ Assertions.assertEquals(
+ getTableLocation("paimon_db.paimon_tbl"),
+ s"${dBLocation.getCanonicalPath}/paimon_tbl")
+ spark.sql("insert into paimon_db.paimon_tbl select '1', 'n', 'cc'")
+
+ spark.sql("alter table paimon_db.paimon_tbl add partition(name='cc', `pt`='aa') ")
+ }
+ }
+ }
+ }
+ }
+
test("Paimon DDL with hive catalog: create database with props") {
Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
catalogName =>