You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by db...@apache.org on 2025/07/02 11:27:26 UTC
(impala) branch master updated (7b25a7b07 -> 191aec529)
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 7b25a7b07 IMPALA-13887: Incorporate column/field information into cache key
new c5072807d IMPALA-12337: Implement delete orphan files for Iceberg table
new bff356157 IMPALA-14194: Fix empty file handling with recent Ozone versions
new 191aec529 IMPALA-14185: Error unnesting nested array from Iceberg with DELETE files
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
common/thrift/JniCatalog.thrift | 8 +
.../AlterTableExecuteExpireSnapshotsStmt.java | 8 +-
.../AlterTableExecuteRemoveOrphanFilesStmt.java | 125 +++++++++++
.../analysis/AlterTableExecuteRollbackStmt.java | 9 +-
.../impala/analysis/AlterTableExecuteStmt.java | 23 +-
.../org/apache/impala/catalog/FileDescriptor.java | 30 ++-
.../iceberg/ImpalaIcebergDeleteOrphanFiles.java | 250 +++++++++++++++++++++
.../java/org/apache/impala/planner/UnnestNode.java | 14 ++
.../apache/impala/service/CatalogOpExecutor.java | 19 ++
.../impala/service/IcebergCatalogOpExecutor.java | 18 ++
.../org/apache/impala/analysis/AnalyzeDDLTest.java | 65 +++++-
.../java/org/apache/impala/analysis/ToSqlTest.java | 22 ++
tests/query_test/test_iceberg.py | 68 ++++++
tests/query_test/test_nested_types.py | 23 ++
14 files changed, 646 insertions(+), 36 deletions(-)
create mode 100644 fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRemoveOrphanFilesStmt.java
create mode 100644 fe/src/main/java/org/apache/impala/catalog/iceberg/ImpalaIcebergDeleteOrphanFiles.java
(impala) 03/03: IMPALA-14185: Error unnesting nested array from Iceberg with DELETE files
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 191aec5298e6f5e2ca352f19b9abb3b111232554
Author: Daniel Becker <db...@apache.org>
AuthorDate: Fri Jun 27 14:43:50 2025 +0200
IMPALA-14185: Error unnesting nested array from Iceberg with DELETE files
When trying to doubly unnest a 2D array from an Iceberg table that has
delete files but not for every data file, we run into an error:
Filtering an unnested collection that comes from a UNION [ALL] is not
supported yet.
This is because there is a UNION node because of the Iceberg delete
files, and there is an added "not-empty" conjunct on the collections.
IMPALA-12753 describes a bug where a conjunct on an unnested collection
coming from a UNION ALL is only applied to the first UNION operand. To
avoid incorrectness, we disabled this case in the commit for
IMPALA-12695, but its unintended consequence is that it leads to this
error with Iceberg tables.
However, in this case with Iceberg deletes, the bug described in
IMPALA-12753 is not present because both sides of the UNION have the
same tuple id, so conjuncts are naturally applied to both sides.
This commit relaxes the check, which now does not fire if all UNION
operands have the same tuple ids.
Testing:
- existing tests related to IMPALA-12753 pass
- added a regression test with an Iceberg table with DELETE files
Change-Id: Ifbc6f580586d4b337f33a2f32052aa07f6fca828
Reviewed-on: http://gerrit.cloudera.org:8080/23107
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/planner/UnnestNode.java | 14 +++++++++++++
tests/query_test/test_nested_types.py | 23 ++++++++++++++++++++++
2 files changed, 37 insertions(+)
diff --git a/fe/src/main/java/org/apache/impala/planner/UnnestNode.java b/fe/src/main/java/org/apache/impala/planner/UnnestNode.java
index 7bebe83ab..a0b6a8e4d 100644
--- a/fe/src/main/java/org/apache/impala/planner/UnnestNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/UnnestNode.java
@@ -90,6 +90,9 @@ public class UnnestNode extends PlanNode {
// Filtering an unnested collection that comes from a UNION [ALL] is not supported, see
// IMPALA-12753.
+ // The exception is if all children of the UNION node have the same tuple id(s), because
+ // then the conjuncts are naturally applied to all UNION operands. This is the case for
+ // UNION nodes inserted because of Iceberg delete operations. See IMPALA-14185.
private void checkUnnestFromUnionWithPredicate(Analyzer analyzer)
throws AnalysisException {
PlanNode subplanInputNode = containingSubplanNode_.getChild(0);
@@ -97,6 +100,8 @@ public class UnnestNode extends PlanNode {
UnionNode union = (UnionNode) subplanInputNode;
+ if (allUnionChildrenHaveSameTupleIds(union)) return;
+
// Tuple descriptors of the UNION and their descendants (for complex types).
List<TupleDescriptor> unionDescs = new ArrayList<>();
for (TupleId tid : union.getTupleIds()) {
@@ -125,6 +130,15 @@ public class UnnestNode extends PlanNode {
}
}
+ private static boolean allUnionChildrenHaveSameTupleIds(UnionNode union) {
+ if (union.getChildren().size() < 2) return true;
+
+ final List<TupleId> firstChildTupleIds = union.getChild(0).getTupleIds();
+ return union.getChildren().stream()
+ .map(planNode -> planNode.getTupleIds())
+ .allMatch(tupleIdList -> tupleIdList.equals(firstChildTupleIds));
+ }
+
// Returns the TupleDescriptors contained by 'tuple' (includes item tuple descs of
// collections).
private void getCollTupleDescs(TupleDescriptor tuple,
diff --git a/tests/query_test/test_nested_types.py b/tests/query_test/test_nested_types.py
index 6d21bac0f..ed4878858 100644
--- a/tests/query_test/test_nested_types.py
+++ b/tests/query_test/test_nested_types.py
@@ -194,6 +194,29 @@ class TestNestedCollectionsInSelectList(ImpalaTestSuite):
"""Queries where a map column is in the select list"""
self.run_test_case('QueryTest/nested-map-in-select-list', vector)
+ @SkipIfFS.hive
+ def test_nested_array_from_iceberg_with_delete(self, unique_database):
+ """Tests that a 2D array can be unnested from an Iceberg table that has delete files
+ but not for all data files. In this case there is a UNION in the plan.
+ Regression test for IMPALA-14185.
+ """
+ tbl_name = unique_database + ".nested_arr_in_iceberg_with_delete"
+ self.execute_query("create table {} (id INT, arr ARRAY<ARRAY<int>>) stored by \
+ iceberg tblproperties('format-version'='2')".format(tbl_name))
+
+ # INSERTs are done in Hive as Impala cannot write complex types.
+ self.run_stmt_in_hive("insert into {} values ( \
+ 1, array(array(1), array(2), array(3), array(4), array(5)))".format(tbl_name))
+ self.run_stmt_in_hive("insert into {} values ( \
+ 2, array(array(1), array(2), array(3), array(4), array(5)))".format(tbl_name))
+ # Impala can delete rows containing complex types.
+ self.execute_query("delete from {} where id=2".format(tbl_name))
+
+ result = self.execute_query_expect_success(self.client,
+ "select id, a1.item.item unnested_item from {0}, {0}.arr a1, a1.item \
+ order by id, unnested_item".format(tbl_name))
+ assert result.data == ['1\t1', '1\t2', '1\t3', '1\t4', '1\t5']
+
class TestMixedCollectionsAndStructsInSelectList(ImpalaTestSuite):
"""Functional tests for the case where collections and structs are embedded into one
(impala) 02/03: IMPALA-14194: Fix empty file handling with recent Ozone versions
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit bff3561573cfad3a08afc93d7442a748b8d5d4fd
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Tue Jul 1 13:37:42 2025 +0200
IMPALA-14194: Fix empty file handling with recent Ozone versions
TestExplainEmptyPartition.test_non_empty_partition_0_rows can fail in
Ozone builds, as for empty files Ozone started to return NULL as
block locations, instead of an empty array of block locations.
Testing
* tested in Ozone build
Change-Id: I067d66c8f90e235e871f16ac3f7faf190ffa13eb
Reviewed-on: http://gerrit.cloudera.org:8080/23108
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Zoltan Borok-Nagy <bo...@cloudera.com>
---
.../org/apache/impala/catalog/FileDescriptor.java | 30 +++++++++++++---------
1 file changed, 18 insertions(+), 12 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/FileDescriptor.java b/fe/src/main/java/org/apache/impala/catalog/FileDescriptor.java
index aa611585f..63a8db098 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileDescriptor.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileDescriptor.java
@@ -131,18 +131,24 @@ public class FileDescriptor implements Comparable<FileDescriptor> {
Reference<Long> numUnknownDiskIds,
String absPath) throws IOException {
FlatBufferBuilder fbb = new FlatBufferBuilder(1);
- int[] fbFileBlockOffsets = new int[blockLocations.length];
- int blockIdx = 0;
- for (BlockLocation loc : blockLocations) {
- if (isEc) {
- fbFileBlockOffsets[blockIdx++] = FileBlock.createFbFileBlock(
- fbb,
- loc.getOffset(),
- loc.getLength(),
- (short) hostIndex.getOrAddIndex(REMOTE_NETWORK_ADDRESS));
- } else {
- fbFileBlockOffsets[blockIdx++] =
- FileBlock.createFbFileBlock(fbb, loc, hostIndex, numUnknownDiskIds);
+ int[] fbFileBlockOffsets;
+ if (blockLocations == null) {
+ Preconditions.checkState(fileStatus.getLen() == 0);
+ fbFileBlockOffsets = new int[0];
+ } else {
+ fbFileBlockOffsets = new int[blockLocations.length];
+ int blockIdx = 0;
+ for (BlockLocation loc : blockLocations) {
+ if (isEc) {
+ fbFileBlockOffsets[blockIdx++] = FileBlock.createFbFileBlock(
+ fbb,
+ loc.getOffset(),
+ loc.getLength(),
+ (short) hostIndex.getOrAddIndex(REMOTE_NETWORK_ADDRESS));
+ } else {
+ fbFileBlockOffsets[blockIdx++] =
+ FileBlock.createFbFileBlock(fbb, loc, hostIndex, numUnknownDiskIds);
+ }
}
}
return new FileDescriptor(createFbFileDesc(fbb, fileStatus, relPath,
(impala) 01/03: IMPALA-12337: Implement delete orphan files for Iceberg table
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dbecker pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c5072807df6ad5adb48b158540259c08d1ce0424
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Tue Jun 17 13:51:10 2025 -0700
IMPALA-12337: Implement delete orphan files for Iceberg table
This patch implements delete orphan files query for Iceberg table.
The following statement becomes available for Iceberg tables:
- ALTER TABLE <tbl> EXECUTE remove_orphan_files(<timestamp>)
The bulk of implementation copies Hive's implementation of
org.apache.iceberg.actions.DeleteOrphanFiles interface (HIVE-27906,
6b2e21a93ef3c1776b689a7953fc59dbf52e4be4), which this patch rename to
ImpalaIcebergDeleteOrphanFiles.java. Upon execute(),
ImpalaIcebergDeleteOrphanFiles class instance will gather all URI of
valid data files and Iceberg metadata files using Iceberg API. These
valid URIs then will be compared to recursive file listing obtained
through Hadoop FileSystem API under table's 'data' and 'metadata'
directory accordingly. Any unmatched URI from FileSystem API listing
that has modification time less than 'olderThanTimestamp' parameter will
then be removed via Iceberg FileIO API of given Iceberg table. Note that
this is a destructive query that will wipe out any files within Iceberg
table's 'data' and 'metadata' directory that is not addressable by any
valid snapshots.
The execution happens in CatalogD via
IcebergCatalogOpExecutor.alterTableExecuteRemoveOrphanFiles(). CatalogD
supplied CatalogOpExecutor.icebergExecutorService_ as executor service
to execute the Iceberg API planFiles and FileIO API for deletion.
Also fixed toSql() implementation for all ALTER TABLE EXECUTE queries.
Testing:
- Add FE and EE tests.
Change-Id: I5979cdf15048d5a2c4784918533f65f32e888de0
Reviewed-on: http://gerrit.cloudera.org:8080/23042
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Zoltan Borok-Nagy <bo...@cloudera.com>
---
common/thrift/JniCatalog.thrift | 8 +
.../AlterTableExecuteExpireSnapshotsStmt.java | 8 +-
.../AlterTableExecuteRemoveOrphanFilesStmt.java | 125 +++++++++++
.../analysis/AlterTableExecuteRollbackStmt.java | 9 +-
.../impala/analysis/AlterTableExecuteStmt.java | 23 +-
.../iceberg/ImpalaIcebergDeleteOrphanFiles.java | 250 +++++++++++++++++++++
.../apache/impala/service/CatalogOpExecutor.java | 19 ++
.../impala/service/IcebergCatalogOpExecutor.java | 18 ++
.../org/apache/impala/analysis/AnalyzeDDLTest.java | 65 +++++-
.../java/org/apache/impala/analysis/ToSqlTest.java | 22 ++
tests/query_test/test_iceberg.py | 68 ++++++
11 files changed, 591 insertions(+), 24 deletions(-)
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 492f2ae4f..ebb699f49 100644
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -444,6 +444,11 @@ struct TAlterTableExecuteRollbackParams {
3: optional i64 snapshot_id
}
+// Parameters for ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES operations.
+struct TAlterTableExecuteRemoveOrphanFilesParams {
+ 1: required i64 older_than_millis
+}
+
// Parameters for ALTER TABLE EXECUTE ... operations.
struct TAlterTableExecuteParams {
// Parameters for ALTER TABLE EXECUTE EXPIRE_SNAPSHOTS
@@ -451,6 +456,9 @@ struct TAlterTableExecuteParams {
// Parameters for ALTER TABLE EXECUTE ROLLBACK
2: optional TAlterTableExecuteRollbackParams execute_rollback_params
+
+ // Parameters for ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES
+ 3: optional TAlterTableExecuteRemoveOrphanFilesParams remove_orphan_files_params
}
// Parameters for all ALTER TABLE commands.
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteExpireSnapshotsStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteExpireSnapshotsStmt.java
index a51621daa..02b9680f6 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteExpireSnapshotsStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteExpireSnapshotsStmt.java
@@ -56,11 +56,6 @@ public class AlterTableExecuteExpireSnapshotsStmt extends AlterTableExecuteStmt
analyzeOlderThan(analyzer);
}
- @Override
- public String toSql(ToSqlOptions options) {
- return fnCallExpr_.toSql();
- }
-
@Override
public TAlterTableParams toThrift() {
TAlterTableParams params = super.toThrift();
@@ -78,7 +73,8 @@ public class AlterTableExecuteExpireSnapshotsStmt extends AlterTableExecuteStmt
Preconditions.checkNotNull(fnParamValue_);
fnParamValue_.analyze(analyzer);
if (!fnParamValue_.isConstant()) {
- throw new AnalysisException(USAGE + " must be a constant expression: " + toSql());
+ throw new AnalysisException(
+ USAGE + " must be a constant expression: " + fnCallExpr_.toSql());
}
if (fnParamValue_.getType().isStringType()) {
fnParamValue_ = new CastExpr(Type.TIMESTAMP, fnParamValue_);
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRemoveOrphanFilesStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRemoveOrphanFilesStmt.java
new file mode 100644
index 000000000..dcfa5340f
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRemoveOrphanFilesStmt.java
@@ -0,0 +1,125 @@
+// 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.impala.analysis;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.Type;
+import org.apache.impala.common.AnalysisException;
+import org.apache.impala.common.InternalException;
+import org.apache.impala.thrift.TAlterTableExecuteParams;
+import org.apache.impala.thrift.TAlterTableExecuteRemoveOrphanFilesParams;
+import org.apache.impala.thrift.TAlterTableParams;
+import org.apache.impala.thrift.TAlterTableType;
+import org.apache.impala.util.ExprUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Represents an ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES(<parameter>) statement.
+ * Valid <parameter> can either be:
+ * - Timestamp expression, such as (NOW() - interval 5 days).
+ * - Timestamp literal, such as '2025-01-01 10:00:00'.
+ * However, user must be careful to not pass NOW() as timestamp expression because
+ * it will remove files of in-progress operations.
+ */
+public class AlterTableExecuteRemoveOrphanFilesStmt extends AlterTableExecuteStmt {
+ public static final String USAGE = "EXECUTE REMOVE_ORPHAN_FILES(<expression>):";
+ private final static Logger LOG =
+ LoggerFactory.getLogger(AlterTableExecuteRemoveOrphanFilesStmt.class);
+
+ public AlterTableExecuteRemoveOrphanFilesStmt(
+ TableName tableName, FunctionCallExpr fnCallExpr) {
+ super(tableName, fnCallExpr);
+ }
+
+ @Override
+ public String getOperation() {
+ return "EXECUTE REMOVE_ORPHAN_FILES";
+ }
+
+ @Override
+ public void analyze(Analyzer analyzer) throws AnalysisException {
+ super.analyze(analyzer);
+ FeTable table = getTargetTable();
+ if (!(table instanceof FeIcebergTable)) {
+ throw new AnalysisException(
+ "ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES is only supported "
+ + "for Iceberg tables: " + table.getTableName());
+ }
+ analyzeFunctionCallExpr(analyzer, USAGE);
+ analyzeParameter(analyzer);
+ }
+
+ private void analyzeParameter(Analyzer analyzer) throws AnalysisException {
+ Preconditions.checkNotNull(fnParamValue_);
+ fnParamValue_.analyze(analyzer);
+
+ if (!fnParamValue_.isConstant()) {
+ throw new AnalysisException(
+ USAGE + " <expression> must be a constant expression: " + fnCallExpr_.toSql());
+ }
+
+ Expr timestampExpr = getParamConvertibleToTimestamp();
+ if (timestampExpr == null) {
+ throw new AnalysisException(USAGE + " <expression> must be a timestamp, but is '"
+ + fnParamValue_.getType() + "': " + fnCallExpr_.toSql());
+ }
+
+ try {
+ olderThanMillis_ =
+ ExprUtil.localTimestampToUnixTimeMicros(analyzer, timestampExpr) / 1000;
+ LOG.debug(USAGE + " millis: " + olderThanMillis_);
+ } catch (InternalException ie) {
+ throw new AnalysisException("An invalid TIMESTAMP expression has been given "
+ + "to " + USAGE + " the expression " + fnParamValue_.toSql()
+ + " cannot be converted to a TIMESTAMP",
+ ie);
+ }
+ }
+
+ /**
+ * If field fnParamValue_ is a Timestamp, or can be cast to a Timestamp,
+ * then return an Expr for the Timestamp.
+ * @return null if the fnParamValue_ cannot be converted to a Timestamp.
+ */
+ private Expr getParamConvertibleToTimestamp() {
+ Preconditions.checkNotNull(fnParamValue_);
+ Expr timestampExpr = fnParamValue_;
+ if (timestampExpr.getType().isStringType()) {
+ timestampExpr = new CastExpr(Type.TIMESTAMP, fnParamValue_);
+ }
+ if (timestampExpr.getType().isTimestamp()) { return timestampExpr; }
+ return null;
+ }
+
+ @Override
+ public TAlterTableParams toThrift() {
+ TAlterTableParams params = super.toThrift();
+ params.setAlter_type(TAlterTableType.EXECUTE);
+ TAlterTableExecuteParams executeParams = new TAlterTableExecuteParams();
+ TAlterTableExecuteRemoveOrphanFilesParams removeOrphanParams =
+ new TAlterTableExecuteRemoveOrphanFilesParams();
+ executeParams.setRemove_orphan_files_params(removeOrphanParams);
+ removeOrphanParams.setOlder_than_millis(olderThanMillis_);
+ params.setSet_execute_params(executeParams);
+ return params;
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRollbackStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRollbackStmt.java
index 0ddb57cbc..550660552 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRollbackStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteRollbackStmt.java
@@ -74,7 +74,7 @@ public class AlterTableExecuteRollbackStmt extends AlterTableExecuteStmt {
if (!fnParamValue_.isConstant()) {
throw new AnalysisException(
- USAGE + " <expression> must be a constant expression: EXECUTE " + toSql());
+ USAGE + " <expression> must be a constant expression: " + fnCallExpr_.toSql());
}
if ((fnParamValue_ instanceof LiteralExpr)
&& (fnParamValue_.getType().isIntegerType())) {
@@ -104,7 +104,7 @@ public class AlterTableExecuteRollbackStmt extends AlterTableExecuteStmt {
} else {
throw new AnalysisException(USAGE
+ " <expression> must be an integer type or a timestamp, but is '"
- + fnParamValue_.getType() + "': EXECUTE " + toSql());
+ + fnParamValue_.getType() + "': " + fnCallExpr_.toSql());
}
}
}
@@ -125,11 +125,6 @@ public class AlterTableExecuteRollbackStmt extends AlterTableExecuteStmt {
return null;
}
- @Override
- public String toSql(ToSqlOptions options) {
- return fnCallExpr_.toSql();
- }
-
@Override
public TAlterTableParams toThrift() {
TAlterTableParams params = super.toThrift();
diff --git a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteStmt.java b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteStmt.java
index 3ae47dcf0..9a65dddcc 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AlterTableExecuteStmt.java
@@ -44,6 +44,17 @@ public class AlterTableExecuteStmt extends AlterTableStmt {
@Override
public String getOperation() { return "EXECUTE"; }
+ @Override
+ public String toSql(ToSqlOptions options) {
+ StringBuilder sb = new StringBuilder("ALTER TABLE ")
+ .append(getDb())
+ .append(".")
+ .append(getTbl())
+ .append(" EXECUTE ")
+ .append(fnCallExpr_.toSql(options));
+ return sb.toString();
+ }
+
/**
* Return an instance of a subclass of AlterTableExecuteStmt that can analyze the
* execute statement for the function call expression in 'expr'.
@@ -57,11 +68,14 @@ public class AlterTableExecuteStmt extends AlterTableStmt {
case "EXPIRE_SNAPSHOTS":
return new AlterTableExecuteExpireSnapshotsStmt(tableName, fnCallExpr);
case "ROLLBACK": return new AlterTableExecuteRollbackStmt(tableName, fnCallExpr);
+ case "REMOVE_ORPHAN_FILES":
+ return new AlterTableExecuteRemoveOrphanFilesStmt(tableName, fnCallExpr);
default:
throw new AnalysisException(String.format("'%s' is not supported by ALTER "
+ "TABLE <table> EXECUTE. Supported operations are: "
+ "EXPIRE_SNAPSHOTS(<expression>), "
- + "ROLLBACK(<expression>).",
+ + "ROLLBACK(<expression>), "
+ + "REMOVE_ORPHAN_FILES(<expression>).",
functionNameOrig));
}
}
@@ -71,10 +85,11 @@ public class AlterTableExecuteStmt extends AlterTableStmt {
// fnCallExpr_ analyzed here manually, because it is not an actual function but a
// catalog operation.
String fnName = fnCallExpr_.getFnName().toString();
- Preconditions.checkState(
- StringUtils.equalsAnyIgnoreCase(fnName, "EXPIRE_SNAPSHOTS", "ROLLBACK"));
+ Preconditions.checkState(StringUtils.equalsAnyIgnoreCase(
+ fnName, "EXPIRE_SNAPSHOTS", "ROLLBACK", "REMOVE_ORPHAN_FILES"));
if (fnCallExpr_.getParams().size() != 1) {
- throw new AnalysisException(usage + " must have one parameter: " + toSql());
+ throw new AnalysisException(
+ usage + " must have one parameter: " + fnCallExpr_.toSql());
}
fnParamValue_ = fnCallExpr_.getParams().exprs().get(0);
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/iceberg/ImpalaIcebergDeleteOrphanFiles.java b/fe/src/main/java/org/apache/impala/catalog/iceberg/ImpalaIcebergDeleteOrphanFiles.java
new file mode 100644
index 000000000..3b38fce2a
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/catalog/iceberg/ImpalaIcebergDeleteOrphanFiles.java
@@ -0,0 +1,250 @@
+/*
+ * 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.impala.catalog.iceberg;
+
+import static org.apache.iceberg.MetadataTableType.ALL_ENTRIES;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.iceberg.DataTask;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.ReachableFileUtil;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.actions.DeleteOrphanFiles;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
+import org.apache.iceberg.util.Tasks;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Implementation of DeleteOrphanFiles for Impala.
+ * This is copied from HiveIcebergDeleteOrphanFiles.java with minimal changes
+ * (HIVE-27906, 6b2e21a93ef3c1776b689a7953fc59dbf52e4be4).
+ *
+ * Upon execute(), this class instance will gather all URI of valid data files
+ * and Iceberg metadata files using Iceberg API. These valid URIs then will be
+ * compared to recursive file listing obtained through Hadoop FileSystem API
+ * under table's 'data' and 'metadata' directory accordingly. Any unmatched URI
+ * from FileSystem API listing that has modification time less than
+ * 'olderThanTimestamp' parameter will then be removed via Iceberg FileIO API
+ * of given Iceberg table.
+ */
+public class ImpalaIcebergDeleteOrphanFiles implements DeleteOrphanFiles {
+ public static final String METADATA_FOLDER_NAME = "metadata";
+ public static final String DATA_FOLDER_NAME = "data";
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ImpalaIcebergDeleteOrphanFiles.class);
+ private String tableLocation;
+ private long olderThanTimestamp =
+ System.currentTimeMillis() - TimeUnit.DAYS.toMillis(3);
+ private Consumer<String> deleteFunc;
+ private ExecutorService deleteExecutorService =
+ MoreExecutors.newDirectExecutorService();
+
+ private final Configuration conf;
+ private final Table table;
+
+ public ImpalaIcebergDeleteOrphanFiles(Configuration conf, Table table) {
+ this.conf = conf;
+ this.table = table;
+ this.deleteFunc = file -> table.io().deleteFile(file);
+ this.tableLocation = table.location();
+ }
+
+ @Override
+ public ImpalaIcebergDeleteOrphanFiles location(String location) {
+ this.tableLocation = location;
+ return this;
+ }
+
+ @Override
+ public ImpalaIcebergDeleteOrphanFiles olderThan(long newOlderThanTimestamp) {
+ this.olderThanTimestamp = newOlderThanTimestamp;
+ return this;
+ }
+
+ // TODO: Implement later, if there is any use case.
+ @Override
+ public ImpalaIcebergDeleteOrphanFiles deleteWith(Consumer<String> newDeleteFunc) {
+ this.deleteFunc = newDeleteFunc;
+ return this;
+ }
+
+ @Override
+ public ImpalaIcebergDeleteOrphanFiles executeDeleteWith(
+ ExecutorService executorService) {
+ this.deleteExecutorService = executorService;
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ LOG.info("Cleaning orphan files for {}", table.name());
+ ImpalaIcebergDeleteOrphanFilesResult result =
+ new ImpalaIcebergDeleteOrphanFilesResult();
+ result.addDeletedFiles(cleanContentFiles(olderThanTimestamp));
+ result.addDeletedFiles(cleanMetadata(olderThanTimestamp));
+
+ LOG.debug("Deleting {} files while cleaning orphan files for {}",
+ result.deletedFiles.size(), table.name());
+ Tasks.foreach(result.deletedFiles)
+ .executeWith(deleteExecutorService)
+ .retry(3)
+ .stopRetryOn(FileNotFoundException.class)
+ .suppressFailureWhenFinished()
+ .onFailure((file, thrown) -> LOG.warn("Delete failed for file: {}", file, thrown))
+ .run(deleteFunc::accept);
+ return result;
+ }
+
+ private Set<String> cleanContentFiles(long lastTime) {
+ Set<String> validFiles =
+ Sets.union(getAllContentFilePath(), getAllStatisticsFilePath(table));
+ LOG.debug("Valid content file for {} are {}", table.name(), validFiles.size());
+ try {
+ Path dataPath = new Path(tableLocation, DATA_FOLDER_NAME);
+ return getFilesToBeDeleted(lastTime, validFiles, dataPath);
+ } catch (IOException ex) { throw new UncheckedIOException(ex); }
+ }
+
+ private Set<String> getAllContentFilePath() {
+ Set<String> validFilesPath = Sets.newHashSet();
+ Table metadataTable = getMetadataTable();
+
+ TableScan tableScan = metadataTable.newScan();
+ CloseableIterable<FileScanTask> manifestFileScanTasks =
+ tableScan.planWith(deleteExecutorService).planFiles();
+ CloseableIterable<StructLike> entries =
+ CloseableIterable.concat(entriesOfManifest(manifestFileScanTasks));
+
+ for (StructLike entry : entries) {
+ StructLike fileRecord = entry.get(4, StructLike.class);
+ String filePath = fileRecord.get(1, String.class);
+ validFilesPath.add(getUriPath(filePath));
+ }
+ return validFilesPath;
+ }
+
+ private Iterable<CloseableIterable<StructLike>> entriesOfManifest(
+ CloseableIterable<FileScanTask> fileScanTasks) {
+ return Iterables.transform(fileScanTasks, task -> {
+ assert task != null;
+ return ((DataTask) task).rows();
+ });
+ }
+
+ private static Set<String> getAllStatisticsFilePath(Table table) {
+ return ReachableFileUtil.statisticsFilesLocations(table)
+ .stream()
+ .map(ImpalaIcebergDeleteOrphanFiles::getUriPath)
+ .collect(Collectors.toSet());
+ }
+
+ protected Set<String> cleanMetadata(long lastTime) {
+ LOG.info("{} start clean metadata files", table.name());
+ try {
+ Set<String> validFiles = getValidMetadataFiles(table);
+ LOG.debug("Valid metadata files for {} are {}", table.name(), validFiles);
+ Path metadataLocation = new Path(tableLocation, METADATA_FOLDER_NAME);
+ return getFilesToBeDeleted(lastTime, validFiles, metadataLocation);
+ } catch (IOException ioe) { throw new UncheckedIOException(ioe); }
+ }
+
+ private Set<String> getFilesToBeDeleted(
+ long lastTime, Set<String> validFiles, Path location) throws IOException {
+ Set<String> filesToDelete = Sets.newHashSet();
+ FileSystem fs = location.getFileSystem(conf);
+ RemoteIterator<LocatedFileStatus> metadataLocations = fs.listFiles(location, true);
+ while (metadataLocations.hasNext()) {
+ LocatedFileStatus metadataFile = metadataLocations.next();
+ if (metadataFile.getModificationTime() < lastTime
+ && !validFiles.contains(getUriPath(metadataFile.getPath().toString()))) {
+ filesToDelete.add(metadataFile.getPath().toString());
+ }
+ }
+ return filesToDelete;
+ }
+
+ private Table getMetadataTable() {
+ return MetadataTableUtils.createMetadataTableInstance(
+ ((HasTableOperations) table).operations(), table.name(),
+ table.name() + "#" + ALL_ENTRIES.name(), ALL_ENTRIES);
+ }
+
+ private static Set<String> getValidMetadataFiles(Table icebergTable) {
+ Set<String> validFiles = Sets.newHashSet();
+ Iterable<Snapshot> snapshots = icebergTable.snapshots();
+ for (Snapshot snapshot : snapshots) {
+ String manifestListLocation = snapshot.manifestListLocation();
+ validFiles.add(getUriPath(manifestListLocation));
+
+ List<ManifestFile> manifestFiles = snapshot.allManifests(icebergTable.io());
+ for (ManifestFile manifestFile : manifestFiles) {
+ validFiles.add(getUriPath(manifestFile.path()));
+ }
+ }
+ Stream
+ .of(ReachableFileUtil.metadataFileLocations(icebergTable, false).stream(),
+ ReachableFileUtil.statisticsFilesLocations(icebergTable).stream(),
+ Stream.of(ReachableFileUtil.versionHintLocation(icebergTable)))
+ .reduce(Stream::concat)
+ .orElse(Stream.empty())
+ .map(ImpalaIcebergDeleteOrphanFiles::getUriPath)
+ .forEach(validFiles::add);
+
+ return validFiles;
+ }
+
+ private static String getUriPath(String path) { return URI.create(path).getPath(); }
+
+ public static class ImpalaIcebergDeleteOrphanFilesResult implements Result {
+ private final Set<String> deletedFiles = Sets.newHashSet();
+
+ @Override
+ public Iterable<String> orphanFileLocations() {
+ return deletedFiles;
+ }
+
+ public void addDeletedFiles(Set<String> files) { this.deletedFiles.addAll(files); }
+ }
+}
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index d83070c35..38e26783f 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -1583,6 +1583,21 @@ public class CatalogOpExecutor {
EventSequence catalogTimeline, InProgressTableModification modification)
throws ImpalaException {
Preconditions.checkState(tbl.isWriteLockedByCurrentThread());
+
+ // Some operation does not require Iceberg transaction and HMS update.
+ if (params.getAlter_type() == TAlterTableType.EXECUTE) {
+ Preconditions.checkState(params.isSetSet_execute_params());
+ TAlterTableExecuteParams setExecuteParams = params.getSet_execute_params();
+ if (setExecuteParams.isSetRemove_orphan_files_params()) {
+ String removeOrphanSummary =
+ IcebergCatalogOpExecutor.alterTableExecuteRemoveOrphanFiles(tbl,
+ setExecuteParams.getRemove_orphan_files_params(),
+ icebergExecutorService_);
+ addSummary(response, removeOrphanSummary);
+ return false;
+ }
+ }
+
boolean needsToUpdateHms = !IcebergUtil.isHiveCatalog(tbl.getMetaStoreTable());
try {
org.apache.iceberg.Transaction iceTxn = IcebergUtil.getIcebergTransaction(tbl);
@@ -1623,6 +1638,10 @@ public class CatalogOpExecutor {
setExecuteParams.getExpire_snapshots_params(),
icebergExecutorService_);
addSummary(response, expireSummary);
+ } else if (setExecuteParams.isSetRemove_orphan_files_params()) {
+ throw new IllegalStateException(
+ "Alter table execute REMOVE_ORPHAN_FILES should not use "
+ + "Iceberg Transaction.");
} else {
// Cannot happen, but throw just in case.
throw new IllegalStateException(
diff --git a/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
index 5faf6aef4..90e159bf9 100644
--- a/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
@@ -49,6 +49,8 @@ import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Term;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.impala.analysis.IcebergPartitionSpec;
+import org.apache.impala.catalog.CatalogException;
+import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.IcebergTable;
import org.apache.impala.catalog.TableLoadingException;
@@ -57,11 +59,13 @@ import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKe
import org.apache.impala.catalog.iceberg.GroupedContentFiles;
import org.apache.impala.catalog.iceberg.IcebergCatalog;
import org.apache.impala.catalog.iceberg.IcebergHiveCatalog;
+import org.apache.impala.catalog.iceberg.ImpalaIcebergDeleteOrphanFiles;
import org.apache.impala.common.ImpalaRuntimeException;
import org.apache.impala.fb.FbIcebergColumnStats;
import org.apache.impala.fb.FbIcebergDataFile;
import org.apache.impala.thrift.TAlterTableDropPartitionParams;
import org.apache.impala.thrift.TAlterTableExecuteExpireSnapshotsParams;
+import org.apache.impala.thrift.TAlterTableExecuteRemoveOrphanFilesParams;
import org.apache.impala.thrift.TAlterTableExecuteRollbackParams;
import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.TIcebergCatalog;
@@ -250,6 +254,20 @@ public class IcebergCatalogOpExecutor {
return "Rollback executed.";
}
+ /**
+ * Executes an ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES.
+ */
+ public static String alterTableExecuteRemoveOrphanFiles(FeIcebergTable tbl,
+ TAlterTableExecuteRemoveOrphanFilesParams params, ExecutorService executors)
+ throws CatalogException {
+ ImpalaIcebergDeleteOrphanFiles deleteOrphan =
+ new ImpalaIcebergDeleteOrphanFiles(FeFsTable.CONF, tbl.getIcebergApiTable())
+ .olderThan(params.older_than_millis)
+ .executeDeleteWith(executors);
+ deleteOrphan.execute();
+ return "Remove orphan files executed.";
+ }
+
/**
* Deletes files related to specific set of partitions
*/
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
index 9cd1eee25..cd8ea9691 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java
@@ -4559,13 +4559,15 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"rollback('2022-01-04 10:00:00');",
"ALTER TABLE EXECUTE ROLLBACK is only supported for Iceberg tables: " +
"functional.alltypes");
- AnalysisError("alter table functional_parquet.iceberg_partitioned execute " +
- "rollback(id);", "EXECUTE ROLLBACK(<expression>): " +
- "<expression> must be a constant expression: EXECUTE rollback(id)");
- AnalysisError("alter table functional_parquet.iceberg_partitioned execute " +
- "rollback(3.14);", "EXECUTE ROLLBACK(<expression>): <expression> " +
- "must be an integer type or a timestamp, but is 'DECIMAL(3,2)': " +
- "EXECUTE rollback(3.14)");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "rollback(id);",
+ "EXECUTE ROLLBACK(<expression>): "
+ + "<expression> must be a constant expression: rollback(id)");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "rollback(3.14);",
+ "EXECUTE ROLLBACK(<expression>): <expression> "
+ + "must be an integer type or a timestamp, but is 'DECIMAL(3,2)': "
+ + "rollback(3.14)");
AnalysisError("alter table functional_parquet.iceberg_partitioned execute " +
"rollback('2021-02-32 15:52:45');", "An invalid TIMESTAMP expression has been " +
"given to EXECUTE ROLLBACK(<expression>): the expression " +
@@ -4583,6 +4585,55 @@ public class AnalyzeDDLTest extends FrontendTestBase {
"'1111' cannot be converted to a TIMESTAMP");
}
+ @Test
+ public void TestAlterExecuteRemoveOrphanFilesSnapshots() {
+ AnalyzesOk("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files(now());");
+ AnalyzesOk("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files(now() - interval 20 years);");
+ AnalyzesOk("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files('2022-01-04 10:00:00');");
+
+ // Negative tests
+ AnalysisError("alter table nodb.alltypes execute "
+ + "remove_orphan_files('2022-01-04 10:00:00');",
+ "Could not resolve table reference: 'nodb.alltypes'");
+ AnalysisError("alter table functional.alltypes execute "
+ + "remove_orphan_files('2022-01-04 10:00:00');",
+ "ALTER TABLE EXECUTE REMOVE_ORPHAN_FILES is only supported for Iceberg tables: "
+ + "functional.alltypes");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files();",
+ "EXECUTE REMOVE_ORPHAN_FILES(<expression>): must have one parameter");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files(id);",
+ "EXECUTE REMOVE_ORPHAN_FILES(<expression>): "
+ + "<expression> must be a constant expression: remove_orphan_files(id)");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files(3.14);",
+ "EXECUTE REMOVE_ORPHAN_FILES(<expression>): <expression> "
+ + "must be a timestamp, but is 'DECIMAL(3,2)': "
+ + "remove_orphan_files(3.14)");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files('2021-02-32 15:52:45');",
+ "An invalid TIMESTAMP expression has been "
+ + "given to EXECUTE REMOVE_ORPHAN_FILES(<expression>): the expression "
+ + "'2021-02-32 15:52:45' cannot be converted to a TIMESTAMP");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files('the beginning');",
+ "An invalid TIMESTAMP expression has been "
+ + "given to EXECUTE REMOVE_ORPHAN_FILES(<expression>): the expression "
+ + "'the beginning' cannot be converted to a TIMESTAMP");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files(1111,2222);",
+ "EXECUTE REMOVE_ORPHAN_FILES(<expression>): must have one parameter");
+ AnalysisError("alter table functional_parquet.iceberg_partitioned execute "
+ + "remove_orphan_files('1111');",
+ "An invalid TIMESTAMP expression has been "
+ + "given to EXECUTE REMOVE_ORPHAN_FILES(<expression>): the expression "
+ + "'1111' cannot be converted to a TIMESTAMP");
+ }
+
private static String buildLongOwnerName() {
StringBuilder comment = new StringBuilder();
for (int i = 0; i < MetaStoreUtil.MAX_OWNER_LENGTH + 5; i++) {
diff --git a/fe/src/test/java/org/apache/impala/analysis/ToSqlTest.java b/fe/src/test/java/org/apache/impala/analysis/ToSqlTest.java
index 138562db4..b4c6ca400 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ToSqlTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ToSqlTest.java
@@ -583,6 +583,28 @@ public class ToSqlTest extends FrontendTestBase {
"SELECT * FROM functional.alltypes CROSS JOIN functional_parquet.alltypes");
}
+ @Test
+ public void TestAlterTableExecute() throws AnalysisException {
+ testToSql(
+ "alter table iceberg_partitioned "
+ + "execute rollback('2022-01-04 10:00:00')",
+ "functional_parquet",
+ "ALTER TABLE functional_parquet.iceberg_partitioned "
+ + "EXECUTE rollback('2022-01-04 10:00:00')");
+ testToSql(
+ "alter table iceberg_partitioned "
+ + "execute expire_snapshots('2022-01-04 10:00:00')",
+ "functional_parquet",
+ "ALTER TABLE functional_parquet.iceberg_partitioned "
+ + "EXECUTE expire_snapshots('2022-01-04 10:00:00')");
+ testToSql(
+ "alter table iceberg_partitioned "
+ + "execute remove_orphan_files('2022-01-04 10:00:00')",
+ "functional_parquet",
+ "ALTER TABLE functional_parquet.iceberg_partitioned "
+ + "EXECUTE remove_orphan_files('2022-01-04 10:00:00')");
+ }
+
@Test
public void TestStructFields() throws AnalysisException {
String[] tables = new String[] { "allcomplextypes", };
diff --git a/tests/query_test/test_iceberg.py b/tests/query_test/test_iceberg.py
index 7ead8d178..316885249 100644
--- a/tests/query_test/test_iceberg.py
+++ b/tests/query_test/test_iceberg.py
@@ -461,6 +461,74 @@ class TestIcebergTable(IcebergTestSuite):
assert expected_text in str(output)
return output
+ def test_execute_remove_orphan_files(self, unique_database):
+ tbl_name = 'tbl_with_orphan_files'
+ db_tbl = unique_database + ".tbl_with_orphan_files"
+ with self.create_impala_client() as impalad_client:
+ impalad_client.execute("create table {0} (i int) stored as iceberg"
+ .format(db_tbl))
+ insert_q = "insert into {0} values ({1})"
+ self.execute_query_expect_success(impalad_client, insert_q.format(db_tbl, 1))
+ self.execute_query_expect_success(impalad_client, insert_q.format(db_tbl, 2))
+ self.execute_query_expect_success(impalad_client, insert_q.format(db_tbl, 3))
+ result = impalad_client.execute('select i from {} order by i'.format(db_tbl))
+ assert result.data == ['1', '2', '3']
+
+ # Add some junk files to data and metadata dir.
+ DATA_PATH = "/test-warehouse/{0}.db/{1}/data/".format(
+ unique_database, tbl_name)
+ METADATA_PATH = "/test-warehouse/{0}.db/{1}/metadata/".format(
+ unique_database, tbl_name)
+ SRC_DIR = os.path.join(
+ os.environ['IMPALA_HOME'],
+ "testdata/data/iceberg_test/iceberg_mixed_file_format_test/{0}/{1}")
+
+ # Copy first set of junk files.
+ file_parq1 = "00000-0-data-gfurnstahl_20220906113044_157fc172-f5d3-4c70-8653-" \
+ "fff150b6136a-job_16619542960420_0002-1-00001.parquet"
+ file_avro1 = "055baf62-de6d-4583-bf21-f187f9482343-m0.avro"
+ self.filesystem_client.copy_from_local(
+ SRC_DIR.format('data', file_parq1), DATA_PATH)
+ self.filesystem_client.copy_from_local(
+ SRC_DIR.format('metadata', file_avro1), METADATA_PATH)
+ assert self.filesystem_client.exists(DATA_PATH + file_parq1)
+ assert self.filesystem_client.exists(METADATA_PATH + file_avro1)
+ # Keep current time.
+ result = impalad_client.execute('select cast(now() as string)')
+ cp1_time = result.data[0]
+ time.sleep(1)
+
+ # Copy second set of junk files.
+ file_parq2 = "00000-0-data-gfurnstahl_20220906114830_907f72c7-36ac-4135-8315-" \
+ "27ff880faff0-job_16619542960420_0004-1-00001.parquet"
+ file_avro2 = "871d1473-8566-46c0-a530-a2256b3f396f-m0.avro"
+ self.filesystem_client.copy_from_local(
+ SRC_DIR.format('data', file_parq2), DATA_PATH)
+ self.filesystem_client.copy_from_local(
+ SRC_DIR.format('metadata', file_avro2), METADATA_PATH)
+ assert self.filesystem_client.exists(DATA_PATH + file_parq2)
+ assert self.filesystem_client.exists(METADATA_PATH + file_avro2)
+
+ # Execute REMOVE_ORPHAN_FILES at specific timestamp.
+ result = impalad_client.execute(
+ "ALTER TABLE {0} EXECUTE REMOVE_ORPHAN_FILES('{1}')".format(db_tbl, cp1_time))
+ assert result.data[0] == 'Remove orphan files executed.'
+ assert not self.filesystem_client.exists(DATA_PATH + file_parq1)
+ assert not self.filesystem_client.exists(METADATA_PATH + file_parq1)
+ assert self.filesystem_client.exists(DATA_PATH + file_parq2)
+ assert self.filesystem_client.exists(METADATA_PATH + file_avro2)
+
+ # Execute REMOVE_ORPHAN_FILES at now().
+ result = impalad_client.execute(
+ "ALTER TABLE {0} EXECUTE REMOVE_ORPHAN_FILES(now())".format(db_tbl))
+ assert result.data[0] == 'Remove orphan files executed.'
+ assert not self.filesystem_client.exists(DATA_PATH + file_parq2)
+ assert not self.filesystem_client.exists(METADATA_PATH + file_parq2)
+
+ # Assert table still queryable.
+ result = impalad_client.execute('select i from {} order by i'.format(db_tbl))
+ assert result.data == ['1', '2', '3']
+
def test_describe_history_params(self, unique_database):
tbl_name = unique_database + ".describe_history"