[core] Add drop partition for flink sql (apache#2167)

Zouxxyy · Nov 14, 2023 · 723d580 · 723d580
1 parent fc32305
commit 723d580
Show file tree

Hide file tree

Showing 13 changed files with 234 additions and 27 deletions.
diff --git a/docs/content/how-to/altering-tables.md b/docs/content/how-to/altering-tables.md
@@ -268,6 +268,40 @@ ALTER TABLE my_table DROP COLUMN c1;
 
 {{< /tabs >}}
 
+
+## Dropping Partitions
+
+The following SQL drops the partitions of the paimon table.
+
+{{< tabs "drop-partitions" >}}
+
+{{< tab "Flink" >}}
+
+For flink sql, you can specify the partial columns of partition columns, and you can also specify multiple partition values at the same time.
+
+```sql
+ALTER TABLE MyTable DROP PARTITION (`id` = 1);
+
+ALTER TABLE MyTable DROP PARTITION (`id` = 1, `name` = 'paimon');
+
+ALTER TABLE MyTable DROP PARTITION (`id` = 1), PARTITION (`id` = 2);
+
+```
+
+{{< /tab >}}
+
+{{< tab "Spark3" >}}
+
+For spark sql, you need to specify all the partition columns.
+
+```sql
+ALTER TABLE MyTable DROP PARTITION (`id` = 1, `name` = 'paimon');
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
 ## Changing Column Nullability
 
 The following SQL changes nullability of column `coupon_info`.

diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/TypeUtils.java
@@ -79,7 +79,7 @@ public static Object castFromCdcValueString(String s, DataType type) {
         return castFromStringInternal(s, type, true);
     }
 
-    private static Object castFromStringInternal(String s, DataType type, boolean isCdcValue) {
+    public static Object castFromStringInternal(String s, DataType type, boolean isCdcValue) {
         BinaryString str = BinaryString.fromString(s);
         switch (type.getTypeRoot()) {
             case CHAR:

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
@@ -23,23 +23,28 @@
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.lineage.LineageMetaFactory;
+import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.Lock;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.AbstractFileStoreTable;
 import org.apache.paimon.table.CatalogEnvironment;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.FileStoreTableFactory;
 import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.system.SystemTableLoader;
 import org.apache.paimon.utils.StringUtils;
 
 import javax.annotation.Nullable;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.options.CatalogOptions.LINEAGE_META;
@@ -101,6 +106,16 @@ public void createDatabase(String name, boolean ignoreIfExists)
         createDatabaseImpl(name);
     }
 
+    @Override
+    public void dropPartition(Identifier identifier, Map<String, String> partitionSpec)
+            throws TableNotExistException {
+        Table table = getTable(identifier);
+        AbstractFileStoreTable fileStoreTable = (AbstractFileStoreTable) table;
+        FileStoreCommit commit = fileStoreTable.store().newCommit(UUID.randomUUID().toString());
+        commit.dropPartitions(
+                Collections.singletonList(partitionSpec), BatchWriteBuilder.COMMIT_IDENTIFIER);
+    }
+
     protected abstract void createDatabaseImpl(String name);
 
     @Override

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -27,6 +27,7 @@
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 /**
@@ -191,6 +192,17 @@ void renameTable(Identifier fromTable, Identifier toTable, boolean ignoreIfNotEx
     void alterTable(Identifier identifier, List<SchemaChange> changes, boolean ignoreIfNotExists)
             throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException;
 
+    /**
+     * Drop the partition of the specify table.
+     *
+     * @param identifier path of the table to drop partition
+     * @param partitions the partition to be deleted
+     * @throws TableNotExistException if the table does not exist
+     * @throws PartitionNotExistException if the partition does not exist
+     */
+    void dropPartition(Identifier identifier, Map<String, String> partitions)
+            throws TableNotExistException, PartitionNotExistException;
+
     /**
      * Modify an existing table from a {@link SchemaChange}.
      *
@@ -323,6 +335,36 @@ public Identifier identifier() {
         }
     }
 
+    /** Exception for trying to operate on a partition that doesn't exist. */
+    class PartitionNotExistException extends Exception {
+
+        private static final String MSG = "Partition %s do not exist in the table %s.";
+
+        private final Identifier identifier;
+
+        private final Map<String, String> partitionSpec;
+
+        public PartitionNotExistException(
+                Identifier identifier, Map<String, String> partitionSpec) {
+            this(identifier, partitionSpec, null);
+        }
+
+        public PartitionNotExistException(
+                Identifier identifier, Map<String, String> partitionSpec, Throwable cause) {
+            super(String.format(MSG, partitionSpec, identifier.getFullName()), cause);
+            this.identifier = identifier;
+            this.partitionSpec = partitionSpec;
+        }
+
+        public Identifier identifier() {
+            return identifier;
+        }
+
+        public Map<String, String> partitionSpec() {
+            return partitionSpec;
+        }
+    }
+
     /** Exception for trying to alter a column that already exists. */
     class ColumnAlreadyExistException extends Exception {
 

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java
@@ -20,8 +20,14 @@
 
 import org.apache.paimon.predicate.Predicate;
 
+import java.util.Map;
+
 /** Inner {@link TableScan} contains filter push down. */
 public interface InnerTableScan extends TableScan {
 
     InnerTableScan withFilter(Predicate predicate);
+
+    default InnerTableScan withPartitionFilter(Map<String, String> partitionSpec) {
+        return this;
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScanImpl.java
@@ -25,6 +25,8 @@
 import org.apache.paimon.table.source.snapshot.StartingScanner;
 import org.apache.paimon.utils.SnapshotManager;
 
+import java.util.Map;
+
 /** {@link TableScan} implementation for batch planning. */
 public class InnerTableScanImpl extends AbstractInnerTableScan {
 
@@ -51,6 +53,12 @@ public InnerTableScan withFilter(Predicate predicate) {
         return this;
     }
 
+    @Override
+    public InnerTableScan withPartitionFilter(Map<String, String> partitionSpec) {
+        snapshotReader.withPartitionFilter(partitionSpec);
+        return this;
+    }
+
     @Override
     public TableScan.Plan plan() {
         if (startingScanner == null) {

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java
@@ -27,6 +27,7 @@
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 /**
  * An interface for building the {@link TableScan} and {@link TableRead}.
@@ -95,6 +96,9 @@ default ReadBuilder withFilter(List<Predicate> predicates) {
      */
     ReadBuilder withFilter(Predicate predicate);
 
+    /** Push partition filter. */
+    ReadBuilder withPartitionFilter(Map<String, String> partitionSpec);
+
     /**
      * Apply projection to the reader.
      *

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
@@ -25,6 +25,7 @@
 import org.apache.paimon.utils.TypeUtils;
 
 import java.util.Arrays;
+import java.util.Map;
 import java.util.Objects;
 
 /** Implementation for {@link ReadBuilder}. */
@@ -36,6 +37,7 @@ public class ReadBuilderImpl implements ReadBuilder {
 
     private Predicate filter;
     private int[][] projection;
+    private Map<String, String> partitionSpec;
 
     public ReadBuilderImpl(InnerTable table) {
         this.table = table;
@@ -60,6 +62,12 @@ public ReadBuilder withFilter(Predicate filter) {
         return this;
     }
 
+    @Override
+    public ReadBuilder withPartitionFilter(Map<String, String> partitionSpec) {
+        this.partitionSpec = partitionSpec;
+        return this;
+    }
+
     @Override
     public ReadBuilder withProjection(int[][] projection) {
         this.projection = projection;
@@ -68,7 +76,7 @@ public ReadBuilder withProjection(int[][] projection) {
 
     @Override
     public TableScan newScan() {
-        return table.newScan().withFilter(filter);
+        return table.newScan().withFilter(filter).withPartitionFilter(partitionSpec);
     }
 
     @Override

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -34,6 +34,7 @@
 import javax.annotation.Nullable;
 
 import java.util.List;
+import java.util.Map;
 
 /** Read splits from specified {@link Snapshot} with given configuration. */
 public interface SnapshotReader {
@@ -50,6 +51,8 @@ public interface SnapshotReader {
 
     SnapshotReader withFilter(Predicate predicate);
 
+    SnapshotReader withPartitionFilter(Map<String, String> partitionSpec);
+
     SnapshotReader withMode(ScanMode scanMode);
 
     SnapshotReader withLevelFilter(Filter<Integer> levelFilter);

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -39,8 +39,10 @@
 import org.apache.paimon.table.source.ScanMode;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.SplitGenerator;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TypeUtils;
 
 import javax.annotation.Nullable;
 
@@ -124,6 +126,30 @@ public SnapshotReader withSnapshot(Snapshot snapshot) {
         return this;
     }
 
+    @Override
+    public SnapshotReader withPartitionFilter(Map<String, String> partitionSpec) {
+        if (partitionSpec != null) {
+            List<String> partitionKeys = tableSchema.partitionKeys();
+            RowType rowType = tableSchema.logicalPartitionType();
+            PredicateBuilder predicateBuilder = new PredicateBuilder(rowType);
+            List<Predicate> partitionFilters =
+                    partitionSpec.entrySet().stream()
+                            .map(
+                                    m -> {
+                                        int index = partitionKeys.indexOf(m.getKey());
+                                        Object value =
+                                                TypeUtils.castFromStringInternal(
+                                                        m.getValue(),
+                                                        rowType.getTypeAt(index),
+                                                        false);
+                                        return predicateBuilder.equal(index, value);
+                                    })
+                            .collect(Collectors.toList());
+            scan.withPartitionFilter(PredicateBuilder.and(partitionFilters));
+        }
+        return this;
+    }
+
     @Override
     public SnapshotReader withFilter(Predicate predicate) {
         List<String> partitionKeys = tableSchema.partitionKeys();

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -227,6 +227,12 @@ public SnapshotReader withFilter(Predicate predicate) {
             return this;
         }
 
+        @Override
+        public SnapshotReader withPartitionFilter(Map<String, String> partitionSpec) {
+            snapshotReader.withPartitionFilter(partitionSpec);
+            return this;
+        }
+
         public SnapshotReader withMode(ScanMode scanMode) {
             snapshotReader.withMode(scanMode);
             return this;