You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by pr...@apache.org on 2025/01/21 08:47:36 UTC
(impala) branch master updated (3118e41c2 -> dc9066e7b)
This is an automated email from the ASF dual-hosted git repository.
prozsa pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 3118e41c2 IMPALA-2945: Account for duplicate keys on multiple nodes preAgg
new d1a0c2ac1 IMPALA-13205: Do not include Iceberg position fields for MERGE statements with INSERT merge clauses
new 5371e0c6d IMPALA-13666: Provide a non-null fileMetadataStats for HdfsPartition
new dc9066e7b IMPALA-13500: (Addendum) Do not assert CatalogD's partition id
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:
.../apache/impala/analysis/IcebergMergeImpl.java | 28 +-
.../org/apache/impala/catalog/HdfsPartition.java | 57 +++-
.../java/org/apache/impala/catalog/HdfsTable.java | 12 +
.../impala/analysis/AnalyzeModifyStmtsTest.java | 6 +
.../org/apache/impala/planner/PlannerTest.java | 4 +
.../PlannerTest/iceberg-merge-insert-only.test | 345 +++++++++++++++++++++
tests/custom_cluster/test_events_custom_configs.py | 4 +-
7 files changed, 440 insertions(+), 16 deletions(-)
create mode 100644 testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge-insert-only.test
(impala) 01/03: IMPALA-13205: Do not include Iceberg position fields for MERGE statements with INSERT merge clauses
Posted by pr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
prozsa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit d1a0c2ac14ae65263691f277be839ef23bf33bf7
Author: Peter Rozsa <pr...@cloudera.com>
AuthorDate: Wed Oct 9 09:52:40 2024 +0200
IMPALA-13205: Do not include Iceberg position fields for MERGE statements with INSERT merge clauses
This patch makes the Iceberg position field inclusion conditional by
including them only if there are UPDATE or DELETE merge clauses that are
listed in a MERGE statement or the target table has existing delete
files. These fields can be omitted when there's no delete file creation
at the sink of the MERGE statement and the table has no existing delete
files.
Additionally, this change disables MERGE for Iceberg target tables
that contain equality delete files, see IMPALA-13674.
Tests:
- iceberg-merge-insert-only planner test added
Change-Id: Ib62c78dab557625fa86988559b3732591755106f
Reviewed-on: http://gerrit.cloudera.org:8080/21931
Reviewed-by: Noemi Pap-Takacs <np...@cloudera.com>
Reviewed-by: Daniel Becker <da...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/analysis/IcebergMergeImpl.java | 28 +-
.../impala/analysis/AnalyzeModifyStmtsTest.java | 6 +
.../org/apache/impala/planner/PlannerTest.java | 4 +
.../PlannerTest/iceberg-merge-insert-only.test | 345 +++++++++++++++++++++
4 files changed, 376 insertions(+), 7 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java b/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java
index 7ee91178e..6c6c00c45 100644
--- a/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java
+++ b/fe/src/main/java/org/apache/impala/analysis/IcebergMergeImpl.java
@@ -127,6 +127,12 @@ public class IcebergMergeImpl implements MergeImpl {
"Unsupported '%s': '%s' for Iceberg table: %s",
TableProperties.MERGE_MODE, modifyWriteMode, icebergTable_.getFullName()));
}
+ if (!icebergTable_.getContentFileStore().getEqualityDeleteFiles().isEmpty()) {
+ throw new AnalysisException(
+ "MERGE statement is not supported for Iceberg tables "
+ + "containing equality deletes.");
+ //TODO: IMPALA-13674
+ }
for (Column column : icebergTable_.getColumns()) {
Path slotPath =
new Path(targetTableRef_.desc_, Collections.singletonList(column.getName()));
@@ -329,13 +335,21 @@ public class IcebergMergeImpl implements MergeImpl {
VirtualColumn.ICEBERG_PARTITION_SERIALIZED.getName())));
}
- List<Expr> positionMetaExpressions = ImmutableList.of(
- new SlotRef(
- ImmutableList.of(targetTableRef_.getUniqueAlias(),
- VirtualColumn.INPUT_FILE_NAME.getName())),
- new SlotRef(
- ImmutableList.of(targetTableRef_.getUniqueAlias(),
- VirtualColumn.FILE_POSITION.getName())));
+ List<Expr> positionMetaExpressions;
+
+ if (mergeStmt_.hasOnlyInsertCases() && icebergTable_.getContentFileStore()
+ .getDataFilesWithDeletes().isEmpty()) {
+ positionMetaExpressions = Collections.emptyList();
+ } else {
+ // DELETE/UPDATE cases require position information to write delete files
+ positionMetaExpressions = ImmutableList.of(
+ new SlotRef(
+ ImmutableList.of(targetTableRef_.getUniqueAlias(),
+ VirtualColumn.INPUT_FILE_NAME.getName())),
+ new SlotRef(
+ ImmutableList.of(targetTableRef_.getUniqueAlias(),
+ VirtualColumn.FILE_POSITION.getName())));
+ }
selectListItems.add(new SelectListItem(rowPresentExpression, ROW_PRESENT));
selectListItems.addAll(
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java
index da0bf9f61..5f6ac3af1 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzeModifyStmtsTest.java
@@ -527,5 +527,11 @@ public class AnalyzeModifyStmtsTest extends AnalyzerTest {
"Target table 'functional_parquet.iceberg_partition_evolution' is incompatible"
+ " with source expressions.\nExpression 's.`month`' (type: INT) is not "
+ "compatible with column 'date_string_col' (type: STRING)");
+ // Target table contains equality delete files
+ AnalysisError("merge into functional_parquet.iceberg_v2_delete_equality t "
+ + "using functional_parquet.iceberg_v2_delete_equality s "
+ + "on t.id = s.id when not matched then insert *",
+ "MERGE statement is not supported for Iceberg tables "
+ + "containing equality deletes.");
}
}
diff --git a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
index 71135ad2a..38fa9536c 100644
--- a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
@@ -1637,5 +1637,9 @@ public class PlannerTest extends PlannerTestBase {
@Test
public void testIcebergMerge() {
runPlannerTestFile("iceberg-merge");
+ TQueryOptions options = new TQueryOptions();
+ runPlannerTestFile("iceberg-merge-insert-only", options,
+ ImmutableSet.of(PlannerTestOption.EXTENDED_EXPLAIN,
+ PlannerTestOption.INCLUDE_EXPLAIN_HEADER));
}
}
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge-insert-only.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge-insert-only.test
new file mode 100644
index 000000000..f36bc7952
--- /dev/null
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-merge-insert-only.test
@@ -0,0 +1,345 @@
+# Merge into an unpartitioned Iceberg table with an insert case
+merge into functional_parquet.iceberg_v2_no_deletes target
+using functional_parquet.iceberg_partitioned source
+on target.i = source.id
+when not matched then insert values (source.id, source.user)
+---- PLAN
+Max Per-Host Resource Reservation: Memory=1.98MB Threads=3
+Per-Host Resource Estimates: Memory=98MB
+WARNING: The following tables are missing relevant table and/or column statistics.
+functional_parquet.iceberg_partitioned, functional_parquet.iceberg_v2_no_deletes
+Analyzed query: SELECT /* +straight_join */ CAST(CAST(CAST(TupleIsNull(0) AS
+TINYINT) AS INT) + CAST(CAST(CAST(TupleIsNull(1) AS TINYINT) AS SMALLINT) *
+CAST(2 AS SMALLINT) AS INT) AS TINYINT) row_present, target.i, target.s,
+source.* FROM functional_parquet.iceberg_v2_no_deletes target FULL OUTER JOIN
+functional_parquet.iceberg_partitioned source ON target.i = source.id
+
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=97.94MB mem-reservation=1.98MB thread-reservation=3
+WRITE TO HDFS [functional_parquet.iceberg_v2_no_deletes, OVERWRITE=false]
+| output exprs: target.i, target.s
+| mem-estimate=1.35KB mem-reservation=0B thread-reservation=0
+|
+03:MERGE
+| CASE 0: NOT MATCHED BY TARGET
+| | result expressions: source.id, source.`user`
+| | type: INSERT
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=0N,1N,2 row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT)
+|
+02:HASH JOIN [FULL OUTER JOIN]
+| hash predicates: target.i = source.id
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=0N,1N row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.iceberg_partitioned source]
+| HDFS partitions=1/1 files=20 size=22.90KB
+| Iceberg snapshot id: 8270633197658268308
+| stored statistics:
+| table: rows=20 size=22.90KB
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=1
+| mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
+| tuple-ids=1 row-size=44B cardinality=20
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [functional_parquet.iceberg_v2_no_deletes target]
+ HDFS partitions=1/1 files=1 size=625B
+ Iceberg snapshot id: 728158873687794725
+ stored statistics:
+ table: rows=3 size=625B
+ columns: unavailable
+ extrapolated-rows=disabled max-scan-range-rows=3
+ mem-estimate=32.00MB mem-reservation=16.00KB thread-reservation=1
+ tuple-ids=0 row-size=16B cardinality=3
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+Max Per-Host Resource Reservation: Memory=1.98MB Threads=5
+Per-Host Resource Estimates: Memory=98MB
+WARNING: The following tables are missing relevant table and/or column statistics.
+functional_parquet.iceberg_partitioned, functional_parquet.iceberg_v2_no_deletes
+Analyzed query: SELECT /* +straight_join */ CAST(CAST(CAST(TupleIsNull(0) AS
+TINYINT) AS INT) + CAST(CAST(CAST(TupleIsNull(1) AS TINYINT) AS SMALLINT) *
+CAST(2 AS SMALLINT) AS INT) AS TINYINT) row_present, target.i, target.s,
+source.* FROM functional_parquet.iceberg_v2_no_deletes target FULL OUTER JOIN
+functional_parquet.iceberg_partitioned source ON target.i = source.id
+
+F02:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=1.97MB mem-reservation=1.94MB thread-reservation=1
+WRITE TO HDFS [functional_parquet.iceberg_v2_no_deletes, OVERWRITE=false]
+| output exprs: target.i, target.s
+| mem-estimate=1.35KB mem-reservation=0B thread-reservation=0
+|
+03:MERGE
+| CASE 0: NOT MATCHED BY TARGET
+| | result expressions: source.id, source.`user`
+| | type: INSERT
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=0N,1N,2 row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT)
+|
+02:HASH JOIN [FULL OUTER JOIN, PARTITIONED]
+| hash predicates: target.i = source.id
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=0N,1N row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--05:EXCHANGE [HASH(source.id)]
+| | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| | tuple-ids=1 row-size=44B cardinality=20
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+| Per-Host Resources: mem-estimate=64.19MB mem-reservation=32.00KB thread-reservation=2
+| 01:SCAN HDFS [functional_parquet.iceberg_partitioned source, RANDOM]
+| HDFS partitions=1/1 files=20 size=22.90KB
+| Iceberg snapshot id: 8270633197658268308
+| stored statistics:
+| table: rows=20 size=22.90KB
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=1
+| mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
+| tuple-ids=1 row-size=44B cardinality=20
+| in pipelines: 01(GETNEXT)
+|
+04:EXCHANGE [HASH(target.i)]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=16B cardinality=3
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+Per-Host Resources: mem-estimate=32.08MB mem-reservation=16.00KB thread-reservation=2
+00:SCAN HDFS [functional_parquet.iceberg_v2_no_deletes target, RANDOM]
+ HDFS partitions=1/1 files=1 size=625B
+ Iceberg snapshot id: 728158873687794725
+ stored statistics:
+ table: rows=3 size=625B
+ columns: unavailable
+ extrapolated-rows=disabled max-scan-range-rows=3
+ mem-estimate=32.00MB mem-reservation=16.00KB thread-reservation=1
+ tuple-ids=0 row-size=16B cardinality=3
+ in pipelines: 00(GETNEXT)
+====
+# Merge into an unpartitioned Iceberg table with multiple insert cases
+merge into functional_parquet.iceberg_v2_no_deletes target
+using functional_parquet.iceberg_partitioned source
+on target.i = source.id
+when not matched and id < 10 then insert values (cast(source.id + 10 as int) , source.user)
+when not matched and id < 20 then insert values (cast(source.id + 20 as int), source.user)
+when not matched then insert values (source.id, source.user)
+---- PLAN
+Max Per-Host Resource Reservation: Memory=1.98MB Threads=3
+Per-Host Resource Estimates: Memory=98MB
+WARNING: The following tables are missing relevant table and/or column statistics.
+functional_parquet.iceberg_partitioned, functional_parquet.iceberg_v2_no_deletes
+Analyzed query: SELECT /* +straight_join */ CAST(CAST(CAST(TupleIsNull(0) AS
+TINYINT) AS INT) + CAST(CAST(CAST(TupleIsNull(1) AS TINYINT) AS SMALLINT) *
+CAST(2 AS SMALLINT) AS INT) AS TINYINT) row_present, target.i, target.s,
+source.* FROM functional_parquet.iceberg_v2_no_deletes target FULL OUTER JOIN
+functional_parquet.iceberg_partitioned source ON target.i = source.id
+
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=97.94MB mem-reservation=1.98MB thread-reservation=3
+WRITE TO HDFS [functional_parquet.iceberg_v2_no_deletes, OVERWRITE=false]
+| output exprs: target.i, target.s
+| mem-estimate=1.35KB mem-reservation=0B thread-reservation=0
+|
+03:MERGE
+| CASE 0: NOT MATCHED BY TARGET
+| | filter predicates: id < CAST(10 AS INT)
+| | result expressions: CAST(CAST(source.id AS BIGINT) + CAST(10 AS BIGINT) AS INT), source.`user`
+| | type: INSERT
+| CASE 1: NOT MATCHED BY TARGET
+| | filter predicates: id < CAST(20 AS INT)
+| | result expressions: CAST(CAST(source.id AS BIGINT) + CAST(20 AS BIGINT) AS INT), source.`user`
+| | type: INSERT
+| CASE 2: NOT MATCHED BY TARGET
+| | result expressions: source.id, source.`user`
+| | type: INSERT
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=0N,1N,2 row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT)
+|
+02:HASH JOIN [FULL OUTER JOIN]
+| hash predicates: target.i = source.id
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=0N,1N row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.iceberg_partitioned source]
+| HDFS partitions=1/1 files=20 size=22.90KB
+| Iceberg snapshot id: 8270633197658268308
+| stored statistics:
+| table: rows=20 size=22.90KB
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=1
+| mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
+| tuple-ids=1 row-size=44B cardinality=20
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [functional_parquet.iceberg_v2_no_deletes target]
+ HDFS partitions=1/1 files=1 size=625B
+ Iceberg snapshot id: 728158873687794725
+ stored statistics:
+ table: rows=3 size=625B
+ columns: unavailable
+ extrapolated-rows=disabled max-scan-range-rows=3
+ mem-estimate=32.00MB mem-reservation=16.00KB thread-reservation=1
+ tuple-ids=0 row-size=16B cardinality=3
+ in pipelines: 00(GETNEXT)
+---- DISTRIBUTEDPLAN
+Max Per-Host Resource Reservation: Memory=1.98MB Threads=5
+Per-Host Resource Estimates: Memory=98MB
+WARNING: The following tables are missing relevant table and/or column statistics.
+functional_parquet.iceberg_partitioned, functional_parquet.iceberg_v2_no_deletes
+Analyzed query: SELECT /* +straight_join */ CAST(CAST(CAST(TupleIsNull(0) AS
+TINYINT) AS INT) + CAST(CAST(CAST(TupleIsNull(1) AS TINYINT) AS SMALLINT) *
+CAST(2 AS SMALLINT) AS INT) AS TINYINT) row_present, target.i, target.s,
+source.* FROM functional_parquet.iceberg_v2_no_deletes target FULL OUTER JOIN
+functional_parquet.iceberg_partitioned source ON target.i = source.id
+
+F02:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=1.97MB mem-reservation=1.94MB thread-reservation=1
+WRITE TO HDFS [functional_parquet.iceberg_v2_no_deletes, OVERWRITE=false]
+| output exprs: target.i, target.s
+| mem-estimate=1.35KB mem-reservation=0B thread-reservation=0
+|
+03:MERGE
+| CASE 0: NOT MATCHED BY TARGET
+| | filter predicates: id < CAST(10 AS INT)
+| | result expressions: CAST(CAST(source.id AS BIGINT) + CAST(10 AS BIGINT) AS INT), source.`user`
+| | type: INSERT
+| CASE 1: NOT MATCHED BY TARGET
+| | filter predicates: id < CAST(20 AS INT)
+| | result expressions: CAST(CAST(source.id AS BIGINT) + CAST(20 AS BIGINT) AS INT), source.`user`
+| | type: INSERT
+| CASE 2: NOT MATCHED BY TARGET
+| | result expressions: source.id, source.`user`
+| | type: INSERT
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=0N,1N,2 row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT)
+|
+02:HASH JOIN [FULL OUTER JOIN, PARTITIONED]
+| hash predicates: target.i = source.id
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=0N,1N row-size=60B cardinality=23
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--05:EXCHANGE [HASH(source.id)]
+| | mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| | tuple-ids=1 row-size=44B cardinality=20
+| | in pipelines: 01(GETNEXT)
+| |
+| F01:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
+| Per-Host Resources: mem-estimate=64.19MB mem-reservation=32.00KB thread-reservation=2
+| 01:SCAN HDFS [functional_parquet.iceberg_partitioned source, RANDOM]
+| HDFS partitions=1/1 files=20 size=22.90KB
+| Iceberg snapshot id: 8270633197658268308
+| stored statistics:
+| table: rows=20 size=22.90KB
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=1
+| mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
+| tuple-ids=1 row-size=44B cardinality=20
+| in pipelines: 01(GETNEXT)
+|
+04:EXCHANGE [HASH(target.i)]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=16B cardinality=3
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+Per-Host Resources: mem-estimate=32.08MB mem-reservation=16.00KB thread-reservation=2
+00:SCAN HDFS [functional_parquet.iceberg_v2_no_deletes target, RANDOM]
+ HDFS partitions=1/1 files=1 size=625B
+ Iceberg snapshot id: 728158873687794725
+ stored statistics:
+ table: rows=3 size=625B
+ columns: unavailable
+ extrapolated-rows=disabled max-scan-range-rows=3
+ mem-estimate=32.00MB mem-reservation=16.00KB thread-reservation=1
+ tuple-ids=0 row-size=16B cardinality=3
+ in pipelines: 00(GETNEXT)
+====
+# Merge into a partitioned Iceberg table with multiple insert cases
+merge into functional_parquet.iceberg_partition_transforms_zorder target
+using (select ts, s, cast(i + 10 as int) i, j from functional_parquet.iceberg_partition_transforms_zorder) source
+on target.i = source.i
+when not matched and source.i < 20 then insert values (source.ts, source.s, source.i, source.j)
+when not matched then insert *
+---- PLAN
+Max Per-Host Resource Reservation: Memory=13.98MB Threads=3
+Per-Host Resource Estimates: Memory=162MB
+WARNING: The following tables are missing relevant table and/or column statistics.
+functional_parquet.iceberg_partition_transforms_zorder
+Analyzed query: SELECT /* +straight_join */ CAST(CAST(CAST(TupleIsNull(0) AS
+TINYINT) AS INT) + CAST(CAST(CAST(TupleIsNull(1) AS TINYINT) AS SMALLINT) *
+CAST(2 AS SMALLINT) AS INT) AS TINYINT) row_present, target.ts, target.s,
+target.i, target.j, target.partition__spec__id,
+target.iceberg__partition__serialized, source.* FROM
+functional_parquet.iceberg_partition_transforms_zorder target FULL OUTER JOIN
+(SELECT ts, s, CAST(CAST(i AS BIGINT) + CAST(10 AS BIGINT) AS INT) i, j FROM
+functional_parquet.iceberg_partition_transforms_zorder) source ON target.i =
+source.i
+
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=161.94MB mem-reservation=13.98MB thread-reservation=3
+WRITE TO HDFS [functional_parquet.iceberg_partition_transforms_zorder, OVERWRITE=false, PARTITION-KEYS=(year(target.ts),iceberg_bucket_transform(target.s, 5))]
+| output exprs: ts, s, i, j
+| mem-estimate=122B mem-reservation=0B thread-reservation=0
+|
+04:SORT
+| order by: LEXICAL: year(target.ts) ASC NULLS LAST, iceberg_bucket_transform(target.s, 5) ASC NULLS LAST, ZORDER: i, j
+| materialized: year(target.ts), iceberg_bucket_transform(target.s, 5)
+| mem-estimate=12.00MB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=6 row-size=61B cardinality=2
+| in pipelines: 04(GETNEXT), 00(OPEN)
+|
+03:MERGE
+| CASE 0: NOT MATCHED BY TARGET
+| | filter predicates: CAST(CAST(i AS BIGINT) + CAST(10 AS BIGINT) AS INT) < CAST(20 AS INT)
+| | result expressions: ts, s, CAST(CAST(i AS BIGINT) + CAST(10 AS BIGINT) AS INT), j
+| | type: INSERT
+| CASE 1: NOT MATCHED BY TARGET
+| | result expressions: ts, s, CAST(CAST(i AS BIGINT) + CAST(10 AS BIGINT) AS INT), j
+| | type: INSERT
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=0N,1N,3 row-size=88B cardinality=2
+| in pipelines: 00(GETNEXT)
+|
+02:HASH JOIN [FULL OUTER JOIN]
+| hash predicates: target.i = CAST(i + 10 AS INT)
+| fk/pk conjuncts: assumed fk/pk
+| mem-estimate=1.94MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=0N,1N row-size=88B cardinality=2
+| in pipelines: 00(GETNEXT), 01(OPEN)
+|
+|--01:SCAN HDFS [functional_parquet.iceberg_partition_transforms_zorder]
+| HDFS partitions=1/1 files=1 size=1.08KB
+| Iceberg snapshot id: 7739113163707801067
+| stored statistics:
+| table: rows=1 size=1.08KB
+| columns: unavailable
+| extrapolated-rows=disabled max-scan-range-rows=1
+| mem-estimate=64.00MB mem-reservation=32.00KB thread-reservation=1
+| tuple-ids=1 row-size=36B cardinality=1
+| in pipelines: 01(GETNEXT)
+|
+00:SCAN HDFS [functional_parquet.iceberg_partition_transforms_zorder target]
+ HDFS partitions=1/1 files=1 size=1.08KB
+ Iceberg snapshot id: 7739113163707801067
+ stored statistics:
+ table: rows=1 size=1.08KB
+ columns missing stats: ts, s, i, j
+ extrapolated-rows=disabled max-scan-range-rows=1
+ mem-estimate=96.00MB mem-reservation=48.00KB thread-reservation=1
+ tuple-ids=0 row-size=52B cardinality=1
+ in pipelines: 00(GETNEXT)
+====
\ No newline at end of file
(impala) 03/03: IMPALA-13500: (Addendum) Do not assert CatalogD's partition id
Posted by pr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
prozsa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit dc9066e7b29208e6dfdf8276a56c9045bd859637
Author: Sai Hemanth Gantasala <sa...@cloudera.com>
AuthorDate: Tue Dec 10 23:02:49 2024 -0800
IMPALA-13500: (Addendum) Do not assert CatalogD's partition id
Amend verifying logs in impala to exclude partition id info.
partition_id assigned by catalogD may not be in serial order, so it is
best to avoid checking partition id in the logs.
Change-Id: I27cdeb2a4bed8afa30a27d05c7399c78af5bcebb
Reviewed-on: http://gerrit.cloudera.org:8080/22198
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/custom_cluster/test_events_custom_configs.py | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/tests/custom_cluster/test_events_custom_configs.py b/tests/custom_cluster/test_events_custom_configs.py
index 1deaf24c8..3b2d60c21 100644
--- a/tests/custom_cluster/test_events_custom_configs.py
+++ b/tests/custom_cluster/test_events_custom_configs.py
@@ -1323,8 +1323,8 @@ class TestEventProcessingCustomConfigs(TestEventProcessingCustomConfigsBase):
"insert into {} partition(p) values (0,0),(1,1),(2,2)".format(test_tbl))
self.client.execute("select * from {}".format(test_tbl))
EventProcessorUtils.wait_for_event_processing(self)
- log_regex = r"Invalidated objects in cache: \[partition %s:p=\d \(id=%%d\)\]" \
- % test_tbl
+ log_regex = r"Invalidated objects in cache: \[partition {}:p={}"\
+ .format(test_tbl, '%d')
self.assert_impalad_log_contains('INFO', log_regex % 0, expected_count=1,
timeout_s=20)
self.assert_impalad_log_contains('INFO', log_regex % 1, expected_count=1,
(impala) 02/03: IMPALA-13666: Provide a non-null fileMetadataStats for HdfsPartition
Posted by pr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
prozsa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 5371e0c6df3e329398712af3ebb739465b947454
Author: Fang-Yu Rao <fa...@cloudera.com>
AuthorDate: Sun Jan 12 21:18:48 2025 -0800
IMPALA-13666: Provide a non-null fileMetadataStats for HdfsPartition
IMPALA-13154 added the method getFileMetadataStats() to
HdfsPartition.java that would return the file metadata statistics. The
method requires the corresponding HdfsPartition instance to have a
non-null field of 'fileMetadataStats_'.
This patch revises two existing constructors of HdfsPartition to provide
a non-null value for 'fileMetadataStats'. This makes it easier for a
third party extension to set up and update the field of
'fileMetadataStats_'. A third party extension has to update the field of
'fileMetadataStats_' if it would like to use this field to get the size
of the partition since all three fields in 'fileMetadataStats_' are
defaulted to 0.
A new constructor was also added for HdfsPartition that allows a third
party extension to provide their own FileMetadataStats when
instantiating an HdfsPartition. To facilitate instantiating a
FileMetadataStats, a new constructor was added for FileMetadataStats
that takes in a List of FileDescriptor's to construct a
FileMetadataStats.
Change-Id: I7e690729fcaebb1e380cc61f2b746783c86dcbf7
Reviewed-on: http://gerrit.cloudera.org:8080/22340
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/catalog/HdfsPartition.java | 57 +++++++++++++++++++---
.../java/org/apache/impala/catalog/HdfsTable.java | 12 +++++
2 files changed, 62 insertions(+), 7 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
index 40978e3d5..b712e9a7b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -767,11 +767,16 @@ public class HdfsPartition extends CatalogObjectImpl
private final long lastRefreshEventId_;
// File statistics corresponding to the encoded file descriptors.
- private FileMetadataStats fileMetadataStats_ = null;
+ private final FileMetadataStats fileMetadataStats_;
/**
- * Constructor. Needed for third party extensions that want to use their own builder
- * to construct the object.
+ * Constructor. Needed for third party extensions that want to use their own builder
+ * to construct the object. A third party extension has to update the field of
+ * 'fileMetadataStats_' if it would like to use this field to get the size of the
+ * partition.
+ *
+ * Note that 'partitionIdCounter_' is not accessible to the subclasses in that its
+ * private.
*/
protected HdfsPartition(HdfsTable table, long prevId, String partName,
List<LiteralExpr> partitionKeyValues, HdfsStorageDescriptor fileFormatDescriptor,
@@ -789,9 +794,18 @@ public class HdfsPartition extends CatalogObjectImpl
isMarkedCached, accessLevel, hmsParameters, cachedMsPartitionDescriptor,
partitionStats, hasIncrementalStats, numRows, writeId,
inFlightEvents, /*createEventId=*/-1L, /*lastCompactionId*/-1L,
- /*lastRefreshEventId*/-1L, /*fileMetadataStats*/null);
+ /*lastRefreshEventId*/-1L, new HdfsTable.FileMetadataStats());
}
+ /**
+ * Constructor. Needed for third party extensions that want to use their own builder
+ * to construct the object. A third party extension has to update the field of
+ * 'fileMetadataStats_' if it would like to use this field to get the size of the
+ * partition.
+ *
+ * Note that 'partitionIdCounter_' is not accessible to the subclasses in that its
+ * private.
+ */
protected HdfsPartition(HdfsTable table, long id, long prevId, String partName,
List<LiteralExpr> partitionKeyValues, HdfsStorageDescriptor fileFormatDescriptor,
@Nonnull ImmutableList<byte[]> encodedFileDescriptors,
@@ -808,7 +822,35 @@ public class HdfsPartition extends CatalogObjectImpl
isMarkedCached, accessLevel, hmsParameters, cachedMsPartitionDescriptor,
partitionStats, hasIncrementalStats, numRows, writeId,
inFlightEvents, /*createEventId=*/-1L, /*lastCompactionId*/-1L,
- /*lastRefreshEventId*/-1L, /*fileMetadataStats*/null);
+ /*lastRefreshEventId*/-1L, new HdfsTable.FileMetadataStats());
+ }
+
+ /**
+ * Constructor. Needed for third party extensions that want to use their own builder
+ * to construct the object. It allows third party extensions to provide their own
+ * FileMetadataStats when instantiating an HdfsPartition.
+ *
+ * Note that 'partitionIdCounter_' is not accessible to the subclasses in that its
+ * private.
+ */
+ protected HdfsPartition(HdfsTable table, long id, long prevId, String partName,
+ List<LiteralExpr> partitionKeyValues, HdfsStorageDescriptor fileFormatDescriptor,
+ @Nonnull ImmutableList<byte[]> encodedFileDescriptors,
+ ImmutableList<byte[]> encodedInsertFileDescriptors,
+ ImmutableList<byte[]> encodedDeleteFileDescriptors,
+ HdfsPartitionLocationCompressor.Location location,
+ boolean isMarkedCached, TAccessLevel accessLevel, Map<String, String> hmsParameters,
+ CachedHmsPartitionDescriptor cachedMsPartitionDescriptor,
+ byte[] partitionStats, boolean hasIncrementalStats, long numRows, long writeId,
+ InFlightEvents inFlightEvents, long createEventId, long lastCompactionId,
+ HdfsTable.FileMetadataStats fileMetadataStats) {
+ this(table, partitionIdCounter_.getAndIncrement(), prevId, partName,
+ partitionKeyValues, fileFormatDescriptor, encodedFileDescriptors,
+ encodedInsertFileDescriptors, encodedDeleteFileDescriptors, location,
+ isMarkedCached, accessLevel, hmsParameters, cachedMsPartitionDescriptor,
+ partitionStats, hasIncrementalStats, numRows, writeId,
+ inFlightEvents, /*createEventId=*/-1L, /*lastCompactionId*/-1L,
+ /*lastRefreshEventId*/-1L, fileMetadataStats);
}
protected HdfsPartition(HdfsTable table, long id, long prevId, String partName,
@@ -822,6 +864,7 @@ public class HdfsPartition extends CatalogObjectImpl
byte[] partitionStats, boolean hasIncrementalStats, long numRows, long writeId,
InFlightEvents inFlightEvents, long createEventId, long lastCompactionId,
long lastRefreshEventId, FileMetadataStats fileMetadataStats) {
+ Preconditions.checkArgument(fileMetadataStats != null);
table_ = table;
id_ = id;
prevId_ = prevId;
@@ -1069,7 +1112,7 @@ public class HdfsPartition extends CatalogObjectImpl
}
public FileMetadataStats getFileMetadataStats() {
- return Preconditions.checkNotNull(fileMetadataStats_);
+ return fileMetadataStats_;
}
@Override
@@ -1177,7 +1220,7 @@ public class HdfsPartition extends CatalogObjectImpl
@Override
public long getSize() {
- if (fileMetadataStats_ != null) {
+ if (fileMetadataStats_.numFiles > 0) {
return fileMetadataStats_.totalFileBytes;
}
long result = 0;
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index e6331fed0..59b12ce75 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -362,6 +362,18 @@ public class HdfsTable extends Table implements FeFsTable {
// Total size (in bytes) of all files in a table/partition.
public long totalFileBytes = 0;
+ public FileMetadataStats() {}
+
+ /**
+ * This constructor allows third party extensions to instantiate a FileMetadataStats
+ * with a List of FileDescriptor's.
+ */
+ public FileMetadataStats(List<FileDescriptor> fds) {
+ for (HdfsPartition.FileDescriptor fd : fds) {
+ accumulate(fd);
+ }
+ }
+
// Initializes the values of the storage stats.
public void init() {
numFiles = 0;