You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by wz...@apache.org on 2024/01/10 00:44:44 UTC
(impala) branch master updated: IMPALA-12605: Fix ALTER TABLE SET PARTITION SPEC field id distribution
This is an automated email from the ASF dual-hosted git repository.
wzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
The following commit(s) were added to refs/heads/master by this push:
new cdac777c5 IMPALA-12605: Fix ALTER TABLE SET PARTITION SPEC field id distribution
cdac777c5 is described below
commit cdac777c51febc99500b8426c2b3aabc7e9addd7
Author: Peter Rozsa <pr...@cloudera.com>
AuthorDate: Fri Dec 15 18:44:58 2023 +0100
IMPALA-12605: Fix ALTER TABLE SET PARTITION SPEC field id distribution
This patch modifies the partition specification updating mechanism to
use an updated Iceberg API. The new API preserves the existing
partition specification by, for instance, generating new field IDs for
modified partitioning terms. Additionally, the new API incorporates
transactional support, as a result, the conditional branching based
on the 'needsTxn' criterion has been eliminated. For Iceberg V1 tables,
the new method correctly adds VOID(*) partition transform for removed
fields at updates, for V2 tables, the updated field could be reused,
but their order remains when they are first introduced in the spec,
these changes are reflected in the test changes.
Tests:
- modified already existing tests
- added e2e tests for V1 and V2 tables
Change-Id: I958107d08e50d7bd9044a57bd2fc02816414012d
Reviewed-on: http://gerrit.cloudera.org:8080/20823
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../impala/analysis/IcebergPartitionExpr.java | 8 ++
.../apache/impala/service/CatalogOpExecutor.java | 25 ++--
.../impala/service/IcebergCatalogOpExecutor.java | 26 ++---
.../java/org/apache/impala/util/IcebergUtil.java | 54 ++++++++-
.../QueryTest/iceberg-delete-partitioned.test | 8 +-
.../queries/QueryTest/iceberg-negative.test | 4 +-
.../QueryTest/iceberg-partitioned-insert.test | 28 ++---
tests/query_test/test_iceberg.py | 128 ++++++++++++++++++++-
8 files changed, 226 insertions(+), 55 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionExpr.java b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionExpr.java
index 20859b25a..c889fb4e5 100644
--- a/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionExpr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergPartitionExpr.java
@@ -19,8 +19,10 @@ package org.apache.impala.analysis;
import java.util.List;
import java.util.Optional;
+import java.util.stream.Collectors;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.transforms.Transforms;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.IcebergColumn;
import org.apache.impala.catalog.Type;
@@ -129,6 +131,12 @@ public class IcebergPartitionExpr extends Expr {
List<PartitionField> partitionFields =
partitionSpec_.getFieldsBySourceId(icebergColumn.getFieldId());
+ // Removing VOID transforms
+ partitionFields = partitionFields.stream().filter(
+ partitionField -> !partitionField.transform()
+ .equals(Transforms.alwaysNull())).collect(
+ Collectors.toList());
+
if (partitionFields.isEmpty()) {
throw new AnalysisException(
"Partition exprs cannot contain non-partition column(s): " + toSql());
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 cd4086584..29178180c 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -86,6 +86,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.mr.Catalogs;
import org.apache.impala.analysis.AlterTableSortByStmt;
import org.apache.impala.analysis.FunctionName;
@@ -1412,7 +1413,6 @@ public class CatalogOpExecutor {
Preconditions.checkState(tbl.isWriteLockedByCurrentThread());
boolean needsToUpdateHms = !isIcebergHmsIntegrationEnabled(tbl.getMetaStoreTable());
try {
- boolean needsTxn = true;
org.apache.iceberg.Transaction iceTxn = IcebergUtil.getIcebergTransaction(tbl);
switch (params.getAlter_type()) {
case ADD_COLUMNS:
@@ -1457,15 +1457,12 @@ public class CatalogOpExecutor {
}
break;
case SET_PARTITION_SPEC:
- // Set partition spec uses 'TableOperations', not transactions.
- needsTxn = false;
// Partition spec is not stored in HMS.
needsToUpdateHms = false;
TAlterTableSetPartitionSpecParams setPartSpecParams =
params.getSet_partition_spec_params();
IcebergCatalogOpExecutor.alterTableSetPartitionSpec(tbl,
- setPartSpecParams.getPartition_spec(),
- catalog_.getCatalogServiceId(), newCatalogVersion);
+ setPartSpecParams.getPartition_spec(), iceTxn);
addSummary(response, "Updated partition spec.");
break;
case SET_TBL_PROPERTIES:
@@ -1492,17 +1489,15 @@ public class CatalogOpExecutor {
"Unsupported ALTER TABLE operation for Iceberg tables: " +
params.getAlter_type());
}
- if (needsTxn) {
- if (!needsToUpdateHms) {
- IcebergCatalogOpExecutor.addCatalogVersionToTxn(iceTxn,
- catalog_.getCatalogServiceId(), newCatalogVersion);
- }
- if (debugAction != null) {
- DebugUtils.executeDebugAction(debugAction, DebugUtils.ICEBERG_COMMIT);
- }
- iceTxn.commitTransaction();
+ if (!needsToUpdateHms) {
+ IcebergCatalogOpExecutor.addCatalogVersionToTxn(iceTxn,
+ catalog_.getCatalogServiceId(), newCatalogVersion);
+ }
+ if (debugAction != null) {
+ DebugUtils.executeDebugAction(debugAction, DebugUtils.ICEBERG_COMMIT);
}
- } catch (IllegalArgumentException ex) {
+ iceTxn.commitTransaction();
+ } catch (IllegalArgumentException | ValidationException ex) {
throw new ImpalaRuntimeException(String.format(
"Failed to ALTER table '%s': %s", params.getTable_name().table_name,
ex.getMessage()));
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 eb385d09b..32e0faff6 100644
--- a/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/IcebergCatalogOpExecutor.java
@@ -36,14 +36,14 @@ import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.RowDelta;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
-import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableOperations;
import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdatePartitionSpec;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.ValidationException;
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.FeIcebergTable;
@@ -178,21 +178,15 @@ public class IcebergCatalogOpExecutor {
* Sets new default partition spec for an Iceberg table.
*/
public static void alterTableSetPartitionSpec(FeIcebergTable feTable,
- TIcebergPartitionSpec partSpec, String catalogServiceId, long catalogVersion)
- throws TableLoadingException, ImpalaRuntimeException {
+ TIcebergPartitionSpec partSpec, Transaction transaction)
+ throws ImpalaRuntimeException {
BaseTable iceTable = (BaseTable)feTable.getIcebergApiTable();
- TableOperations tableOp = iceTable.operations();
- TableMetadata metadata = tableOp.current();
- Schema schema = metadata.schema();
- PartitionSpec newPartSpec = IcebergUtil.createIcebergPartition(schema, partSpec);
- TableMetadata newMetadata = metadata.updatePartitionSpec(newPartSpec);
- Map<String, String> properties = new HashMap<>(newMetadata.properties());
- properties.put(MetastoreEventPropertyKey.CATALOG_SERVICE_ID.getKey(),
- catalogServiceId);
- properties.put(MetastoreEventPropertyKey.CATALOG_VERSION.getKey(),
- String.valueOf(catalogVersion));
- newMetadata = newMetadata.replaceProperties(properties);
- tableOp.commit(metadata, newMetadata);
+ UpdatePartitionSpec updatePartitionSpec = transaction.updateSpec();
+ iceTable.spec().fields().forEach(partitionField -> updatePartitionSpec.removeField(
+ partitionField.name()));
+ List<Term> partitioningTerms = IcebergUtil.getPartitioningTerms(partSpec);
+ partitioningTerms.forEach(updatePartitionSpec::addField);
+ updatePartitionSpec.commit();
}
/**
diff --git a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
index d86205a64..f0dba7713 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -57,11 +57,14 @@ import org.apache.iceberg.TableScan;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.expressions.Term;
import org.apache.iceberg.hadoop.HadoopFileIO;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.mr.Catalogs;
import org.apache.iceberg.transforms.PartitionSpecVisitor;
+import org.apache.iceberg.transforms.Transforms;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
@@ -244,13 +247,60 @@ public class IcebergUtil {
} else if (transformType == TIcebergPartitionTransformType.VOID) {
builder.alwaysNull(partitionField.getOrig_field_name());
} else {
- throw new ImpalaRuntimeException(String.format("Skip partition: %s, %s",
- partitionField.getOrig_field_name(), transformType));
+ throw new ImpalaRuntimeException(
+ String.format("Unknown partition transform '%s' for field '%s",
+ transformType, partitionField.getOrig_field_name()));
}
}
return builder.build();
}
+ /**
+ * Transforms TIcebergPartitionSpec to a list of Iceberg terms.
+ * @param partSpec The partition spec Thrift object
+ * @return List of Iceberg terms.
+ * @throws ImpalaRuntimeException When the transform type is not supported
+ */
+ public static List<Term> getPartitioningTerms(TIcebergPartitionSpec partSpec)
+ throws ImpalaRuntimeException {
+ List<Term> result = new ArrayList<>();
+ for (TIcebergPartitionField field : partSpec.getPartition_fields()) {
+ result.add(getPartitioningTerm(field));
+ }
+ return result;
+ }
+
+ /**
+ * Retrieves the Iceberg partitioning term for a given partition field.
+ * @param field partition field for which the partitioning term is generated.
+ * @return The Iceberg partitioning term as a Term.
+ * @throws ImpalaRuntimeException When the transform type is not supported
+ * @see TIcebergPartitionField
+ * @see Expressions
+ * @see Transforms
+ */
+ private static Term getPartitioningTerm(TIcebergPartitionField field)
+ throws ImpalaRuntimeException {
+ TIcebergPartitionTransformType transformType = field.getTransform()
+ .getTransform_type();
+ String fieldName = field.getOrig_field_name();
+ int transformParam = field.getTransform().getTransform_param();
+ switch (transformType) {
+ case IDENTITY: return Expressions.transform(fieldName, Transforms.identity());
+ case HOUR: return Expressions.hour(fieldName);
+ case DAY: return Expressions.day(fieldName);
+ case MONTH: return Expressions.month(fieldName);
+ case YEAR: return Expressions.year(fieldName);
+ case BUCKET: return Expressions.bucket(fieldName, transformParam);
+ case TRUNCATE: return Expressions.truncate(fieldName, transformParam);
+ case VOID: return Expressions.transform(fieldName, Transforms.alwaysNull());
+ default:
+ throw new ImpalaRuntimeException(
+ String.format("Unknown partition transform '%s' for field '%s",
+ transformType, fieldName));
+ }
+ }
+
/**
* Returns true if 'msTable' uses HiveCatalog.
*/
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
index 7d4b999b9..f2809539b 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-delete-partitioned.test
@@ -368,10 +368,10 @@ row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/i=10/delete-.*
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/i=10/(?!delete-).*.parq','.*B','','.*'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/i=20/delete-.*parq','.*B','','.*'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/i=20/(?!delete-).*.parq','.*B','','.*'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc=f/delete-.*parq','.*B','','.*'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc=f/delete-.*parq','.*B','','.*'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc=f/(?!delete-).*.parq','.*B','','.*'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc=t/(?!delete-).*.parq','.*B','','.*'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc_1=f/delete-.*parq','.*B','','.*'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc_1=f/delete-.*parq','.*B','','.*'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc_1=f/(?!delete-).*.parq','.*B','','.*'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/evolve_part/data/s_trunc_1=t/(?!delete-).*.parq','.*B','','.*'
---- TYPES
STRING, STRING, STRING, STRING
====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
index ba26d193a..6d7cba6f2 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-negative.test
@@ -686,12 +686,12 @@ AnalysisException: Source column 's' in table $DATABASE.iceberg_alter_part must
---- QUERY
ALTER TABLE iceberg_alter_part SET PARTITION SPEC (YEAR(i));
---- CATCH
-ImpalaRuntimeException: Failed to ALTER table 'iceberg_alter_part': Cannot partition type int by year
+ImpalaRuntimeException: Failed to ALTER table 'iceberg_alter_part': Cannot bind: year cannot transform int values from 'i'
====
---- QUERY
ALTER TABLE iceberg_alter_part SET PARTITION SPEC (HOUR(d));
---- CATCH
-ImpalaRuntimeException: Failed to ALTER table 'iceberg_alter_part': Cannot partition type date by hour
+ImpalaRuntimeException: Failed to ALTER table 'iceberg_alter_part': Cannot bind: hour cannot transform date values from 'd'
====
---- QUERY
CREATE TABLE clone_ice LIKE functional_parquet.alltypestiny STORED AS ICEBERG;
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-partitioned-insert.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-partitioned-insert.test
index b1806a66e..e54a8f6af 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-partitioned-insert.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-partitioned-insert.test
@@ -503,18 +503,18 @@ Path,Size,Partition,EC Policy
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/[^=]*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/i=3/d=2020-12-09/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/i=4/d=2020-12-10/.*.0.parq','.*','','$ERASURECODE_POLICY'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/d_year=2020/i=5/s_bucket=2/.*.0.parq','.*','','$ERASURECODE_POLICY'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/d_year=2020/i=6/s_bucket=0/.*.0.parq','.*','','$ERASURECODE_POLICY'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/i=5/d_year=2020/s_bucket_5=2/.*.0.parq','.*','','$ERASURECODE_POLICY'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_alter_part/data/i=6/d_year=2020/s_bucket_5=0/.*.0.parq','.*','','$ERASURECODE_POLICY'
---- TYPES
STRING, STRING, STRING, STRING
====
---- QUERY
SHOW PARTITIONS ice_alter_part;
---- RESULTS
-'{"d_year":"50","i":"5","s_bucket":"2"}',1,1
-'{"d_year":"50","i":"6","s_bucket":"0"}',1,1
'{"i":"3","d":"18605"}',1,1
'{"i":"4","d":"18606"}',1,1
+'{"i":"5","d":null,"d_year":"50","s_bucket_5":"2"}',1,1
+'{"i":"6","d":null,"d_year":"50","s_bucket_5":"0"}',1,1
'{}',2,1
---- TYPES
STRING, BIGINT, BIGINT
@@ -577,17 +577,17 @@ Path,Size,Partition,EC Policy
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=o/d_year=2001/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2002/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2003/.*.0.parq','.*','','$ERASURECODE_POLICY'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/i=4/d_year=2004/.*.0.parq','.*','','$ERASURECODE_POLICY'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/d_year=2004/i=4/.*.0.parq','.*','','$ERASURECODE_POLICY'
---- TYPES
STRING, STRING, STRING, STRING
====
---- QUERY
SHOW PARTITIONS ice_void;
---- RESULTS
-'{"i":"4","s_null":null,"d_year":"34"}',1,1
'{"i_null":null,"s_trunc":"o","d_year":"31"}',2,1
'{"i_null":null,"s_trunc":"t","d_year":"32"}',1,1
'{"i_null":null,"s_trunc":"t","d_year":"33"}',1,1
+'{"i_null":null,"s_trunc":null,"d_year":"34","i":"4","s_null":null}',1,1
---- TYPES
STRING, BIGINT, BIGINT
====
@@ -614,7 +614,7 @@ Path,Size,Partition,EC Policy
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=o/d_year=2001/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2002/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2003/.*.0.parq','.*','','$ERASURECODE_POLICY'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/i=4/d_year=2004/.*.0.parq','.*','','$ERASURECODE_POLICY'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/d_year=2004/i=4/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/i=5/[^=/]*.0.parq','.*','','$ERASURECODE_POLICY'
---- TYPES
STRING, STRING, STRING, STRING
@@ -622,11 +622,11 @@ STRING, STRING, STRING, STRING
---- QUERY
SHOW PARTITIONS ice_void;
---- RESULTS
-'{"i":"4","s_null":null,"d_year":"34"}',1,1
-'{"i":"5","s_null":null,"d_null":null}',2,1
'{"i_null":null,"s_trunc":"o","d_year":"31"}',2,1
'{"i_null":null,"s_trunc":"t","d_year":"32"}',1,1
'{"i_null":null,"s_trunc":"t","d_year":"33"}',1,1
+'{"i_null":null,"s_trunc":null,"d_year":"34","i":"4","s_null":null}',1,1
+'{"i_null":null,"s_trunc":null,"d_year":null,"i":"5","s_null":null,"d_null":null}',2,1
---- TYPES
STRING, BIGINT, BIGINT
====
@@ -655,7 +655,7 @@ Path,Size,Partition,EC Policy
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=o/d_year=2001/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2002/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/s_trunc=t/d_year=2003/.*.0.parq','.*','','$ERASURECODE_POLICY'
-row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/i=4/d_year=2004/.*.0.parq','.*','','$ERASURECODE_POLICY'
+row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/d_year=2004/i=4/.*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/i=5/[^=/]*.0.parq','.*','','$ERASURECODE_POLICY'
row_regex:'$NAMENODE/test-warehouse/$DATABASE.db/ice_void/data/[^=/]*.0.parq','.*','','$ERASURECODE_POLICY'
---- TYPES
@@ -664,12 +664,12 @@ STRING, STRING, STRING, STRING
---- QUERY
SHOW PARTITIONS ice_void;
---- RESULTS
-'{"i":"4","s_null":null,"d_year":"34"}',1,1
-'{"i":"5","s_null":null,"d_null":null}',2,1
-'{"i_null":null,"s_null":null,"d_null":null}',2,1
'{"i_null":null,"s_trunc":"o","d_year":"31"}',2,1
'{"i_null":null,"s_trunc":"t","d_year":"32"}',1,1
'{"i_null":null,"s_trunc":"t","d_year":"33"}',1,1
+'{"i_null":null,"s_trunc":null,"d_year":"34","i":"4","s_null":null}',1,1
+'{"i_null":null,"s_trunc":null,"d_year":null,"i":"5","s_null":null,"d_null":null}',2,1
+'{"i_null":null,"s_trunc":null,"d_year":null,"i":null,"s_null":null,"d_null":null}',2,1
---- TYPES
STRING, BIGINT, BIGINT
====
@@ -796,7 +796,7 @@ show partitions special_char_partitions;
'{"i":"3","s":"11=14=31","s_trunc":"11=1"}',1,1
'{"i":"4","s":"","s_trunc":""}',1,1
'{"i":"5","s":null,"s_trunc":null}',1,1
-'{"s2":"98\\/22"}',1,1
+'{"i":null,"s":null,"s_trunc":null,"s2":"98\\/22"}',1,1
---- TYPES
STRING,BIGINT,BIGINT
====
diff --git a/tests/query_test/test_iceberg.py b/tests/query_test/test_iceberg.py
index a505dfea3..dd87a006b 100644
--- a/tests/query_test/test_iceberg.py
+++ b/tests/query_test/test_iceberg.py
@@ -27,7 +27,7 @@ import random
import re
import time
-from subprocess import check_call
+from subprocess import check_call, check_output
# noinspection PyUnresolvedReferences
from parquet.ttypes import ConvertedType
@@ -667,7 +667,7 @@ class TestIcebergTable(IcebergTestSuite):
os.remove(local_file)
- # Get hdfs path to manifest list that belongs to the sanpshot identified by
+ # Get hdfs path to manifest list that belongs to the snapshot identified by
# 'snapshot_counter'.
def get_manifest_list_hdfs_path(self, tmp_path_prefix, db_name, table_name,
snapshot_counter):
@@ -725,6 +725,130 @@ class TestIcebergTable(IcebergTestSuite):
os.remove(local_path)
return datafiles
+ def get_latest_metadata_path(self, database_name, table_name):
+ describe = 'describe extended %s.%s ' % (database_name, table_name)
+ output = self.client.execute(describe)
+ metadata_location = [s for s in output.data if s.startswith('\tmetadata_location')]
+ assert len(metadata_location) == 1
+ metadata_location_split = metadata_location[0].split('\t')
+ assert len(metadata_location_split) == 3
+ metadata_path = metadata_location_split[2]
+ return metadata_path
+
+ # Get the current partition spec as JSON from the latest metadata file
+ def get_current_partition_spec(self, database_name, table_name):
+ hdfs_path = self.get_latest_metadata_path(database_name, table_name)
+ output = check_output(['hadoop', 'fs', '-cat', hdfs_path])
+
+ current_partition_spec = None
+ metadata = json.loads(output)
+
+ current_spec_id = metadata['default-spec-id']
+ for spec in metadata['partition-specs']:
+ if spec['spec-id'] == current_spec_id:
+ current_partition_spec = spec
+ break
+ return current_partition_spec
+
+ @SkipIf.not_dfs
+ def test_partition_spec_update_v1(self, vector, unique_database):
+ # Create table
+ table_name = "ice_part"
+ qualified_table_name = "%s.%s" % (unique_database, table_name)
+ create_table = 'create table %s ' \
+ '(s string, i int) partitioned by spec(truncate(5, s), identity(i)) ' \
+ 'stored as iceberg' \
+ % qualified_table_name
+ self.client.execute(create_table)
+
+ partition_spec = self.get_current_partition_spec(unique_database, table_name)
+ assert len(partition_spec['fields']) == 2
+ truncate_s = partition_spec['fields'][0]
+ identity_i = partition_spec['fields'][1]
+ # At table creation, partition names does not contain the parameter value.
+ assert truncate_s['name'] == 's_trunc'
+ assert identity_i['name'] == 'i'
+ assert truncate_s['field-id'] == 1000
+ assert identity_i['field-id'] == 1001
+
+ # Partition evolution
+ partition_evolution = 'alter table %s set partition ' \
+ 'spec(identity(i), truncate(6,s))' % qualified_table_name
+ self.client.execute(partition_evolution)
+
+ # V1 partition evolution keeps the old, modified fields, but changes their
+ # transform to VOID.
+ evolved_partition_spec = self.get_current_partition_spec(unique_database, table_name)
+ assert len(evolved_partition_spec['fields']) == 3
+ old_truncate_s = evolved_partition_spec['fields'][0]
+ identity_i = evolved_partition_spec['fields'][1]
+ truncate_s = evolved_partition_spec['fields'][2]
+ assert old_truncate_s['name'] == 's_trunc'
+ assert identity_i['name'] == 'i'
+ # Modified field name contains the parameter value.
+ assert truncate_s['name'] == 's_trunc_6'
+ assert old_truncate_s['field-id'] == 1000
+ assert identity_i['field-id'] == 1001
+ # field-id increases for the modified field.
+ assert truncate_s['field-id'] == 1002
+ assert old_truncate_s['transform'] == 'void'
+
+ @SkipIf.not_dfs
+ def test_partition_spec_update_v2(self, vector, unique_database):
+ # Create table
+ table_name = "ice_part"
+ qualified_table_name = "%s.%s" % (unique_database, table_name)
+ create_table = 'create table %s ' \
+ '(s string, i int) partitioned by spec(truncate(5, s), identity(i)) ' \
+ 'stored as iceberg tblproperties ("format-version" = "2")' \
+ % qualified_table_name
+ self.client.execute(create_table)
+
+ partition_spec = self.get_current_partition_spec(unique_database, table_name)
+ assert len(partition_spec['fields']) == 2
+ truncate_s = partition_spec['fields'][0]
+ identity_i = partition_spec['fields'][1]
+ # At table creation, partition names does not contain the parameter value.
+ assert truncate_s['name'] == 's_trunc'
+ assert identity_i['name'] == 'i'
+ assert truncate_s['field-id'] == 1000
+ assert identity_i['field-id'] == 1001
+
+ # Partition evolution for s
+ partition_evolution = 'alter table %s set partition ' \
+ 'spec(identity(i), truncate(6,s))' % qualified_table_name
+ self.client.execute(partition_evolution)
+
+ # V2 partition evolution updates the previous partitioning for
+ # the modified field.
+ evolved_partition_spec = self.get_current_partition_spec(unique_database, table_name)
+ assert len(evolved_partition_spec['fields']) == 2
+ identity_i = evolved_partition_spec['fields'][0]
+ truncate_s = evolved_partition_spec['fields'][1]
+ assert identity_i['name'] == 'i'
+ # Modified field name contains the parameter value.
+ assert truncate_s['name'] == 's_trunc_6'
+ assert identity_i['field-id'] == 1001
+ # field-id increased for the modified field.
+ assert truncate_s['field-id'] == 1002
+
+ # Partition evolution for i and s
+ partition_evolution = 'alter table %s set partition ' \
+ 'spec(bucket(4, i), truncate(3,s))' \
+ % qualified_table_name
+ self.client.execute(partition_evolution)
+
+ evolved_partition_spec = self.get_current_partition_spec(unique_database, table_name)
+ assert len(evolved_partition_spec['fields']) == 2
+ bucket_i = evolved_partition_spec['fields'][0]
+ truncate_s = evolved_partition_spec['fields'][1]
+ # The field naming follows the transform changes.
+ assert bucket_i['name'] == 'i_bucket_4'
+ assert truncate_s['name'] == 's_trunc_3'
+ # field-id's are increasing with each modification
+ assert bucket_i['field-id'] == 1003
+ assert truncate_s['field-id'] == 1004
+
@SkipIf.not_dfs
def test_writing_metrics_to_metadata(self, vector, unique_database):
# Create table