Support deleting whole partitions in Iceberg table

presto-docs/src/main/sphinx/connector/iceberg.rst

@@ -511,6 +511,25 @@ dropping the table from the metadata catalog using ``TRUNCATE TABLE``.
     -----------+------+-----------+---------
     (0 rows)
 
+DELETE
+^^^^^^^^
+
+The iceberg connector can delete data in one or more entire partitions from tables by using ``DELETE FROM``. For example, to delete from the table ``lineitem``::
+
+     DELETE FROM lineitem;
+
+     DELETE FROM lineitem WHERE linenumber = 1;
+
+     DELETE FROM lineitem WHERE linenumber not in (1, 3, 5, 7) and linestatus in ('O', 'F');
+
+.. note::
+
+    Columns in the filter must all be identity transformed partition columns of the target table.
+
+    Filtered columns only support comparison operators, such as EQUALS, LESS THAN, or LESS THAN EQUALS.
+
+    Deletes must only occur on the latest snapshot.
+
 DROP TABLE
 ^^^^^^^^^^^
 

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergAbstractMetadata.java

@@ -49,11 +49,14 @@
 import org.apache.iceberg.DeleteFiles;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.PartitionSpecParser;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
 import org.apache.iceberg.Transaction;
+import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
@@ -65,11 +68,14 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 import static com.facebook.presto.common.type.BigintType.BIGINT;
+import static com.facebook.presto.iceberg.ExpressionConverter.toIcebergExpression;
 import static com.facebook.presto.iceberg.IcebergColumnHandle.primitiveIcebergColumnHandle;
 import static com.facebook.presto.iceberg.IcebergErrorCode.ICEBERG_INVALID_SNAPSHOT_ID;
 import static com.facebook.presto.iceberg.IcebergTableProperties.FILE_FORMAT_PROPERTY;
@@ -89,6 +95,7 @@
 import static com.google.common.base.Verify.verify;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.lang.String.format;
 import static java.util.Collections.singletonList;
 import static java.util.Objects.requireNonNull;
@@ -414,6 +421,7 @@ public IcebergTableHandle getTableHandle(ConnectorSession session, SchemaTableNa
                 name.getTableName(),
                 name.getTableType(),
                 resolveSnapshotIdByName(table, name),
+                name.getSnapshotId().isPresent(),
                 TupleDomain.all());
     }
 
@@ -438,6 +446,101 @@ public Optional<SystemTable> getSystemTable(ConnectorSession session, SchemaTabl
         return getIcebergSystemTable(tableName, icebergTable);
     }
 
+    @Override
+    public void truncateTable(ConnectorSession session, ConnectorTableHandle tableHandle)
+    {
+        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+        Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
+        try (CloseableIterable<FileScanTask> files = icebergTable.newScan().planFiles()) {
+            removeScanFiles(icebergTable, files);
+        }
+        catch (IOException e) {
+            throw new PrestoException(GENERIC_INTERNAL_ERROR, "failed to scan files for delete", e);
+        }
+    }
+
+    @Override
+    public ConnectorTableHandle beginDelete(ConnectorSession session, ConnectorTableHandle tableHandle)
+    {
+        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+        if (handle.isSnapshotSpecified()) {
+            throw new PrestoException(NOT_SUPPORTED, "This connector do not allow delete data at specified snapshot");
+        }
+        throw new PrestoException(NOT_SUPPORTED, "This connector only supports delete where one or more partitions are deleted entirely");
+    }
+
+    @Override
+    public boolean supportsMetadataDelete(ConnectorSession session, ConnectorTableHandle tableHandle, Optional<ConnectorTableLayoutHandle> tableLayoutHandle)
+    {
+        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+        if (handle.isSnapshotSpecified()) {
+            return false;
+        }
+
+        if (!tableLayoutHandle.isPresent()) {
+            return true;
+        }
+
+        // Allow metadata delete for range filters on partition columns.
+        IcebergTableLayoutHandle layoutHandle = (IcebergTableLayoutHandle) tableLayoutHandle.get();
+
+        TupleDomain<ColumnHandle> domainPredicate = layoutHandle.getTupleDomain();
+        if (domainPredicate.isAll()) {
+            return true;
+        }
+
+        Set<Integer> predicateColumnIds = domainPredicate.getDomains().get().keySet().stream()
+                .map(IcebergColumnHandle.class::cast)
+                .map(IcebergColumnHandle::getId)
+                .collect(toImmutableSet());
+        Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
+
+        boolean supportsMetadataDelete = true;
+        for (PartitionSpec spec : icebergTable.specs().values()) {
+            // Currently we do not support delete when any partition columns in predicate is not transform by identity()
+            Set<Integer> partitionColumnSourceIds = spec.fields().stream()
+                    .filter(field -> field.transform().isIdentity())
+                    .map(PartitionField::sourceId)
+                    .collect(Collectors.toSet());
+
+            if (!partitionColumnSourceIds.containsAll(predicateColumnIds)) {
+                supportsMetadataDelete = false;
+                break;
+            }
+        }
+
+        return supportsMetadataDelete;
+    }
+
+    @Override
+    public OptionalLong metadataDelete(ConnectorSession session, ConnectorTableHandle tableHandle, ConnectorTableLayoutHandle tableLayoutHandle)
+    {
+        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
+        IcebergTableLayoutHandle layoutHandle = (IcebergTableLayoutHandle) tableLayoutHandle;
+
+        Table icebergTable;
+        try {
+            icebergTable = getIcebergTable(session, handle.getSchemaTableName());
+        }
+        catch (Exception e) {
+            throw new TableNotFoundException(handle.getSchemaTableName());
+        }
+
+        TableScan scan = icebergTable.newScan();
+        TupleDomain<ColumnHandle> domainPredicate = layoutHandle.getTupleDomain();
+        if (!domainPredicate.isAll()) {
+            Expression filterExpression = toIcebergExpression(handle.getPredicate());
+            scan = scan.filter(filterExpression);
+        }
+
+        try (CloseableIterable<FileScanTask> files = scan.planFiles()) {
+            return OptionalLong.of(removeScanFiles(icebergTable, files));
+        }
+        catch (IOException e) {
+            throw new PrestoException(GENERIC_INTERNAL_ERROR, "failed to scan files for delete", e);
+        }
+    }
+
     /**
      * Deletes all the files within a particular scan
      *
@@ -456,17 +559,4 @@ private long removeScanFiles(Table icebergTable, Iterable<FileScanTask> scan)
         transaction.commitTransaction();
         return rowsDeleted.get();
     }
-
-    @Override
-    public void truncateTable(ConnectorSession session, ConnectorTableHandle tableHandle)
-    {
-        IcebergTableHandle handle = (IcebergTableHandle) tableHandle;
-        Table icebergTable = getIcebergTable(session, handle.getSchemaTableName());
-        try (CloseableIterable<FileScanTask> files = icebergTable.newScan().planFiles()) {
-            removeScanFiles(icebergTable, files);
-        }
-        catch (IOException e) {
-            throw new PrestoException(GENERIC_INTERNAL_ERROR, "failed to scan files for delete", e);
-        }
-    }
 }

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergTableHandle.java

@@ -33,19 +33,22 @@ public class IcebergTableHandle
     private final TableType tableType;
     private final Optional<Long> snapshotId;
     private final TupleDomain<IcebergColumnHandle> predicate;
+    private final boolean snapshotSpecified;
 
     @JsonCreator
     public IcebergTableHandle(
             @JsonProperty("schemaName") String schemaName,
             @JsonProperty("tableName") String tableName,
             @JsonProperty("tableType") TableType tableType,
             @JsonProperty("snapshotId") Optional<Long> snapshotId,
+            @JsonProperty("snapshotSpecified") boolean snapshotSpecified,
             @JsonProperty("predicate") TupleDomain<IcebergColumnHandle> predicate)
     {
         this.schemaName = requireNonNull(schemaName, "schemaName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.tableType = requireNonNull(tableType, "tableType is null");
         this.snapshotId = requireNonNull(snapshotId, "snapshotId is null");
+        this.snapshotSpecified = requireNonNull(snapshotSpecified, "specifiedSnapshot is null");
         this.predicate = requireNonNull(predicate, "predicate is null");
     }
 
@@ -73,6 +76,12 @@ public Optional<Long> getSnapshotId()
         return snapshotId;
     }
 
+    @JsonProperty
+    public boolean isSnapshotSpecified()
+    {
+        return snapshotSpecified;
+    }
+
     @JsonProperty
     public TupleDomain<IcebergColumnHandle> getPredicate()
     {
@@ -104,13 +113,14 @@ public boolean equals(Object o)
                 Objects.equals(tableName, that.tableName) &&
                 tableType == that.tableType &&
                 Objects.equals(snapshotId, that.snapshotId) &&
+                snapshotSpecified == that.snapshotSpecified &&
                 Objects.equals(predicate, that.predicate);
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(schemaName, tableName, tableType, snapshotId, predicate);
+        return Objects.hash(schemaName, tableName, tableType, snapshotId, snapshotSpecified, predicate);
     }
 
     @Override

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java

@@ -23,6 +23,7 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.SchemaTableName;
+import com.facebook.presto.spi.connector.ConnectorMetadata;
 import com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.FileFormat;
@@ -57,6 +58,7 @@
 import static com.facebook.presto.iceberg.IcebergSessionProperties.isMergeOnReadModeEnabled;
 import static com.facebook.presto.iceberg.util.IcebergPrestoModelConverters.toIcebergTableIdentifier;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.Lists.reverse;
@@ -87,6 +89,13 @@ public static boolean isIcebergTable(com.facebook.presto.hive.metastore.Table ta
         return ICEBERG_TABLE_TYPE_VALUE.equalsIgnoreCase(table.getParameters().get(TABLE_TYPE_PROP));
     }
 
+    public static Table getIcebergTable(ConnectorMetadata metadata, ConnectorSession session, SchemaTableName table)
+    {
+        checkArgument(metadata instanceof IcebergAbstractMetadata, "metadata must be instance of IcebergAbstractMetadata!");
+        IcebergAbstractMetadata icebergMetadata = (IcebergAbstractMetadata) metadata;
+        return icebergMetadata.getIcebergTable(session, table);
+    }
+
     public static Table getHiveIcebergTable(ExtendedHiveMetastore metastore, HdfsEnvironment hdfsEnvironment, ConnectorSession session, SchemaTableName table)
     {
         HdfsContext hdfsContext = new HdfsContext(session, table.getSchemaName(), table.getTableName());