You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2026/10/21 07:54:41 UTC
(impala) branch master updated (867c237b5d -> c8b4c3664c)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 867c237b5d IMPALA-14510: Disable ineffective runtime filters more aggressively
new 0207eaa0a2 IMPALA-15175: Support HBO for AnalyticEvalNode cardinality
new c8b4c3664c IMPALA-15495: Enable UPDATE on Iceberg REST catalog tables
The 2 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:
docs/topics/impala_iceberg_rest_catalog.xml | 4 +-
.../org/apache/impala/analysis/AnalyticExpr.java | 9 +-
.../java/org/apache/impala/analysis/Analyzer.java | 29 +-
.../org/apache/impala/analysis/DeleteStmt.java | 3 +
.../org/apache/impala/analysis/ModifyStmt.java | 4 +-
.../org/apache/impala/analysis/UpdateStmt.java | 3 +
.../org/apache/impala/catalog/FeIcebergTable.java | 6 +-
.../org/apache/impala/catalog/IcebergTable.java | 11 +-
.../org/apache/impala/catalog/VirtualColumn.java | 18 +
.../impala/catalog/local/IcebergMetaProvider.java | 11 +-
.../apache/impala/planner/AnalyticEvalNode.java | 68 ++++
.../apache/impala/service/IcebergDmlFinalizer.java | 9 +-
.../impala/analysis/AnalyzerCapabilityTest.java | 8 +-
.../apache/impala/planner/HboKeyStringTest.java | 217 +++++++++++
.../service/IcebergRestDmlOperationTest.java | 16 +-
.../queries/QueryTest/hbo-analytic-agg.test | 67 ++++
.../queries/QueryTest/hbo-analytic-join.test | 229 +++++++++++
.../QueryTest/iceberg-rest-update-partitions.test | 243 ++++++++++++
.../queries/QueryTest/iceberg-rest-update.test | 427 +++++++++++++++++++++
tests/common/iceberg_rest_test_tables.py | 131 +++++++
tests/custom_cluster/test_iceberg_rest_catalog.py | 132 +++++--
tests/query_test/test_hbo.py | 28 ++
22 files changed, 1590 insertions(+), 83 deletions(-)
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-agg.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-join.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update-partitions.test
create mode 100644 testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update.test
create mode 100644 tests/common/iceberg_rest_test_tables.py
(impala) 02/02: IMPALA-15495: Enable UPDATE on Iceberg REST catalog tables
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c8b4c3664ce40f792aa33b4cacb4c4b15e151924
Author: Nandor Kollar <ko...@gmail.com>
AuthorDate: Thu Oct 15 10:01:11 2026 +0200
IMPALA-15495: Enable UPDATE on Iceberg REST catalog tables
The REST finalization infrastructure (IMPALA-15310) was built to be
operation-agnostic, but UPDATE was gated by two explicit policy checks.
This patch opens those gates while keeping DELETE, MERGE and OPTIMIZE
blocked.
Analysis-time gate (Analyzer.java): ensureTableWriteSupported() now
accepts a boolean indicating whether the caller is a supported REST
DML operation. ModifyStmt gets it from a new abstract method,
isRestDmlSupported(), which UpdateStmt overrides to return true and
DeleteStmt to return false. MergeStmt and
OptimizeStmt continue to call checkTableCapability(WRITE) which
rejects all non-INSERT DML on REST tables. The method
supportsInsert() is renamed to supportsDml() on FeIcebergTable
since it now gates all supported REST DML, not just INSERT. The helper
ensureRestCatalogInsertEnabled() is renamed to
ensureRestCatalogDmlEnabled().
Commit-time gate (IcebergDmlFinalizer.java): validateRestDmlOperation()
uses a SUPPORTED_REST_OPS EnumSet to accept both INSERT and UPDATE.
Row-lineage virtual columns (IcebergMetaProvider.java): REST tables
never exposed _row_id and _last_updated_sequence_number, because
IcebergMetaProvider kept its own hard-coded list of Iceberg virtual
columns, which missed the format-version 3 additions that IcebergTable
has. Both now use VirtualColumn.getIcebergVirtualColumns(formatVersion),
backed by immutable ICEBERG_VIRTUAL_COLUMNS and
ICEBERG_V3_VIRTUAL_COLUMNS lists, so the two paths cannot drift again.
This also fixes SELECT of these columns on V3 REST tables.
No backend changes are needed: the C++ sinks are catalog-agnostic,
the coordinator-side finalization routing is based on rest_catalog_name
(not operation type), and IcebergCatalogOpExecutor.updateRows() already
uses the pure Iceberg API (txn.newRowDelta()).
Testing:
- AnalyzerCapabilityTest updated for renamed methods and new messages.
- IcebergRestDmlOperationTest updated: UPDATE validation passes and
DELETE is rejected.
- test_rest_catalog_insert_into: UPDATE removed from the unsupported
list.
- New test_rest_catalog_update: runs iceberg-rest-update.test, then
injects CommitStateUnknownException (files are kept) and
CommitFailedException (written files are cleaned up) for UPDATE.
- New iceberg-rest-update.test adapted from iceberg-v3-update.test:
basic UPDATE, multi-row, _row_id and _last_updated_sequence_number
verification, DV file checks, single-column, all-types (individual
and multi-column), JOIN-based UPDATE FROM, duplicate error, NULL
handling, column swap, cross-column expression, and
SKIP_UNNEEDED_UPDATES_COL_LIMIT.
- New iceberg-rest-update-partitions.test adapted from
iceberg-v3-update-partitions.test: identity, truncate,
bucket+truncate, date year/month/day, timestamp year/month/day/hour,
and numeric truncate partition transforms. Partition evolution tests
are omitted as ALTER TABLE DDL is not available on REST-backed tables.
- Tables are created via the REST API since DDL is not supported.
Change-Id: I847f051e9a350b730935326c983c7459215c54e2
Assisted-by: Claude Opus 4.6 (Claude Code)
Reviewed-on: http://gerrit.cloudera.org:8080/25046
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Peter Rozsa <pr...@cloudera.com>
---
docs/topics/impala_iceberg_rest_catalog.xml | 4 +-
.../java/org/apache/impala/analysis/Analyzer.java | 29 +-
.../org/apache/impala/analysis/DeleteStmt.java | 3 +
.../org/apache/impala/analysis/ModifyStmt.java | 4 +-
.../org/apache/impala/analysis/UpdateStmt.java | 3 +
.../org/apache/impala/catalog/FeIcebergTable.java | 6 +-
.../org/apache/impala/catalog/IcebergTable.java | 11 +-
.../org/apache/impala/catalog/VirtualColumn.java | 18 +
.../impala/catalog/local/IcebergMetaProvider.java | 11 +-
.../apache/impala/service/IcebergDmlFinalizer.java | 9 +-
.../impala/analysis/AnalyzerCapabilityTest.java | 8 +-
.../service/IcebergRestDmlOperationTest.java | 16 +-
.../QueryTest/iceberg-rest-update-partitions.test | 243 ++++++++++++
.../queries/QueryTest/iceberg-rest-update.test | 427 +++++++++++++++++++++
tests/common/iceberg_rest_test_tables.py | 131 +++++++
tests/custom_cluster/test_iceberg_rest_catalog.py | 132 +++++--
16 files changed, 973 insertions(+), 82 deletions(-)
diff --git a/docs/topics/impala_iceberg_rest_catalog.xml b/docs/topics/impala_iceberg_rest_catalog.xml
index 1137277796..007b743951 100644
--- a/docs/topics/impala_iceberg_rest_catalog.xml
+++ b/docs/topics/impala_iceberg_rest_catalog.xml
@@ -274,11 +274,11 @@ iceberg.rest-catalog.uri=http://rest-catalog-server:9084</codeblock>
be set to <codeph>false</codeph>. Only flat namespace hierarchies are supported.
</li>
<li>
- <b>Write Operations:</b> Only INSERT INTO and INSERT OVERWRITE are
+ <b>Write Operations:</b> Only INSERT INTO, INSERT OVERWRITE and UPDATE are
supported, and only for a catalog configured with a unique
<codeph>iceberg.rest-catalog.name</codeph>. INSERT OVERWRITE replaces only
the partitions that receive new rows, or the whole table if it is
- unpartitioned. UPDATE, DELETE, MERGE, OPTIMIZE and TRUNCATE are not
+ unpartitioned. DELETE, MERGE, OPTIMIZE and TRUNCATE are not
supported.
</li>
<li>
diff --git a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
index 5cd9f00761..bc0e62b7b1 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Analyzer.java
@@ -208,10 +208,10 @@ public class Analyzer {
private static final String TABLE_NOT_SUPPORTED =
"%s not supported. Table %s access type is: %s";
private static final String REST_TABLE_WRITE_NOT_SUPPORTED =
- "Only INSERT INTO and INSERT OVERWRITE are supported for REST-backed Iceberg " +
- "table: %s";
+ "Only INSERT INTO, INSERT OVERWRITE and UPDATE are supported for REST-backed " +
+ "Iceberg table: %s";
private static final String REST_TABLE_UUID_UNAVAILABLE =
- "INSERT is disabled because the Iceberg table UUID is unavailable: %s";
+ "DML is disabled because the Iceberg table UUID is unavailable: %s";
private final static Logger LOG = LoggerFactory.getLogger(Analyzer.class);
@@ -406,9 +406,9 @@ public class Analyzer {
}
}
- static boolean supportsInsert(FeTable table) {
+ static boolean supportsDml(FeTable table) {
return table instanceof FeIcebergTable
- && ((FeIcebergTable)table).supportsInsert();
+ && ((FeIcebergTable)table).supportsDml();
}
static boolean hasProviderDerivedCapabilities(FeTable table) {
@@ -417,20 +417,21 @@ public class Analyzer {
}
/**
- * Check whether a table supports this write operation. 'isInsert' is true for
- * INSERT INTO and INSERT OVERWRITE, and false for every other write, UPSERT included.
+ * Check whether a table supports this write operation. 'isRestDmlSupported' is true
+ * for INSERT INTO, INSERT OVERWRITE and UPDATE, and false for every other write,
+ * UPSERT included.
*/
- static void ensureTableWriteSupported(FeTable table, boolean isInsert)
- throws AnalysisException {
+ static void ensureTableWriteSupported(FeTable table,
+ boolean isRestDmlSupported) throws AnalysisException {
if (!hasProviderDerivedCapabilities(table)) {
ensureTableWriteSupported(table);
return;
}
- if (!isInsert) {
+ if (!isRestDmlSupported) {
throw new AnalysisException(
String.format(REST_TABLE_WRITE_NOT_SUPPORTED, table.getFullName()));
}
- ensureRestCatalogInsertEnabled(table);
+ ensureRestCatalogDmlEnabled(table);
// The provider-derived capability replaces only the HMS access-type check. Keep
// the structural checks that apply to every INSERT target.
ensureTableNotBucketed(table);
@@ -439,11 +440,11 @@ public class Analyzer {
}
}
- private static void ensureRestCatalogInsertEnabled(FeTable table)
+ private static void ensureRestCatalogDmlEnabled(FeTable table)
throws AnalysisException {
- if (!supportsInsert(table)) {
+ if (!supportsDml(table)) {
throw new AnalysisException(String.format(
- "INSERT is disabled because the REST catalog does not have a "
+ "DML is disabled because the REST catalog does not have a "
+ "unique name: %s",
table.getFullName()));
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java b/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
index 947a4fcd37..671b9d48a4 100644
--- a/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/DeleteStmt.java
@@ -49,6 +49,9 @@ public class DeleteStmt extends ModifyStmt {
new ArrayList<>(), other.wherePredicate_.clone());
}
+ @Override
+ protected boolean isRestDmlSupported() { return false; }
+
@Override
protected void createModifyImpl() {
if (table_ instanceof FeKuduTable) {
diff --git a/fe/src/main/java/org/apache/impala/analysis/ModifyStmt.java b/fe/src/main/java/org/apache/impala/analysis/ModifyStmt.java
index 246ad938f8..d23c03e5a9 100644
--- a/fe/src/main/java/org/apache/impala/analysis/ModifyStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/ModifyStmt.java
@@ -95,6 +95,8 @@ public abstract class ModifyStmt extends DmlStatementBase {
wherePredicate_ = wherePredicate;
}
+ protected abstract boolean isRestDmlSupported();
+
@Override
public void collectTableRefs(List<TableRef> tblRefs) {
tblRefs.add(new TableRef(targetTablePath_, null));
@@ -157,7 +159,7 @@ public abstract class ModifyStmt extends DmlStatementBase {
"but the following table is neither: %s",
dstTbl.getFullName()));
}
- Analyzer.checkTableCapability(dstTbl, Analyzer.OperationType.WRITE);
+ Analyzer.ensureTableWriteSupported(dstTbl, isRestDmlSupported());
if (dstTbl instanceof FeIcebergTable) {
setMaxTableSinks(analyzer_.getQueryOptions().getMax_fs_writers());
}
diff --git a/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java b/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
index 693d4d45c0..0be72381e9 100644
--- a/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/UpdateStmt.java
@@ -56,6 +56,9 @@ public class UpdateStmt extends ModifyStmt {
new ArrayList<>(), other.wherePredicate_);
}
+ @Override
+ protected boolean isRestDmlSupported() { return true; }
+
@Override
protected void createModifyImpl() {
// Currently Kudu and Iceberg tables are supported.
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
index 6a7ef78e58..a30e8851d3 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -125,10 +125,10 @@ public interface FeIcebergTable extends FeFsTable {
default boolean hasProviderDerivedCapabilities() { return false; }
/**
- * Whether INSERT INTO and INSERT OVERWRITE can be finalized by the metadata provider.
- * Metadata writes are intentionally excluded.
+ * Whether INSERT INTO, INSERT OVERWRITE and UPDATE can be finalized by the metadata
+ * provider. Metadata writes are intentionally excluded.
*/
- default boolean supportsInsert() {
+ default boolean supportsDml() {
return hasProviderDerivedCapabilities() && getIcebergDmlCatalogName() != null;
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
index b9248eb4d4..881b2dc517 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
@@ -821,15 +821,8 @@ public class IcebergTable extends Table implements FeIcebergTable {
}
private void addVirtualColumns() {
- addVirtualColumn(VirtualColumn.INPUT_FILE_NAME);
- addVirtualColumn(VirtualColumn.FILE_POSITION);
- addVirtualColumn(VirtualColumn.PARTITION_SPEC_ID);
- addVirtualColumn(VirtualColumn.ICEBERG_PARTITION_SERIALIZED);
- addVirtualColumn(VirtualColumn.ICEBERG_DATA_SEQUENCE_NUMBER);
- addVirtualColumn(VirtualColumn.ICEBERG_FIRST_ROW_ID);
- if (getFormatVersion() >= 3) {
- addVirtualColumn(VirtualColumn.ICEBERG_ROW_ID);
- addVirtualColumn(VirtualColumn.ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER);
+ for (VirtualColumn col : VirtualColumn.getIcebergVirtualColumns(getFormatVersion())) {
+ addVirtualColumn(col);
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/VirtualColumn.java b/fe/src/main/java/org/apache/impala/catalog/VirtualColumn.java
index 641976e30f..f5705f1f1b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/VirtualColumn.java
+++ b/fe/src/main/java/org/apache/impala/catalog/VirtualColumn.java
@@ -17,10 +17,13 @@
package org.apache.impala.catalog;
+import java.util.List;
+
import org.apache.impala.thrift.TColumn;
import org.apache.impala.thrift.TVirtualColumnType;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
/**
* Virtual columns are columns that are not stored by the table, but they can reveal some
@@ -71,6 +74,21 @@ public class VirtualColumn extends Column {
public static VirtualColumn BUCKET_ID = new VirtualColumn("BUCKET_ID",
Type.INT, TVirtualColumnType.BUCKET_ID);
+ // Virtual columns of Iceberg V1 and V2 tables.
+ private static final List<VirtualColumn> ICEBERG_BASE_VIRTUAL_COLUMNS =
+ ImmutableList.of(INPUT_FILE_NAME, FILE_POSITION, PARTITION_SPEC_ID,
+ ICEBERG_PARTITION_SERIALIZED, ICEBERG_DATA_SEQUENCE_NUMBER,
+ ICEBERG_FIRST_ROW_ID);
+ // V3 tables add the row lineage columns on top of the base ones.
+ private static final List<VirtualColumn> ICEBERG_V3_VIRTUAL_COLUMNS =
+ ImmutableList.<VirtualColumn>builder()
+ .addAll(ICEBERG_BASE_VIRTUAL_COLUMNS)
+ .add(ICEBERG_ROW_ID, ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER)
+ .build();
+
+ public static List<VirtualColumn> getIcebergVirtualColumns(int formatVersion) {
+ return formatVersion >= 3 ? ICEBERG_V3_VIRTUAL_COLUMNS : ICEBERG_BASE_VIRTUAL_COLUMNS;
+ }
public static VirtualColumn getVirtualColumn(TVirtualColumnType virtColType) {
switch (virtColType) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/IcebergMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/IcebergMetaProvider.java
index a6c076c584..4117e19db7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/IcebergMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/IcebergMetaProvider.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.catalog.Namespace;
@@ -520,14 +521,8 @@ public class IcebergMetaProvider implements MetaProvider {
@Override
public List<VirtualColumn> getVirtualColumns() {
- List<VirtualColumn> ret = new ArrayList<>();
- ret.add(VirtualColumn.INPUT_FILE_NAME);
- ret.add(VirtualColumn.FILE_POSITION);
- ret.add(VirtualColumn.PARTITION_SPEC_ID);
- ret.add(VirtualColumn.ICEBERG_PARTITION_SERIALIZED);
- ret.add(VirtualColumn.ICEBERG_DATA_SEQUENCE_NUMBER);
- ret.add(VirtualColumn.ICEBERG_FIRST_ROW_ID);
- return ret;
+ return VirtualColumn.getIcebergVirtualColumns(
+ ((BaseTable) iceApiTbl_).operations().current().formatVersion());
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/service/IcebergDmlFinalizer.java b/fe/src/main/java/org/apache/impala/service/IcebergDmlFinalizer.java
index 59a533a0b2..203da5307c 100644
--- a/fe/src/main/java/org/apache/impala/service/IcebergDmlFinalizer.java
+++ b/fe/src/main/java/org/apache/impala/service/IcebergDmlFinalizer.java
@@ -17,6 +17,8 @@
package org.apache.impala.service;
+import java.util.EnumSet;
+
import org.apache.iceberg.Transaction;
import org.apache.iceberg.exceptions.CommitStateUnknownException;
import org.apache.impala.catalog.FeIcebergTable;
@@ -36,6 +38,9 @@ import org.slf4j.LoggerFactory;
final class IcebergDmlFinalizer {
private static final Logger LOG = LoggerFactory.getLogger(IcebergDmlFinalizer.class);
+ private static final EnumSet<TIcebergOperation> SUPPORTED_REST_OPS =
+ EnumSet.of(TIcebergOperation.INSERT, TIcebergOperation.UPDATE);
+
@FunctionalInterface
interface PostOperationHook {
void run() throws Exception;
@@ -96,9 +101,9 @@ final class IcebergDmlFinalizer {
static void validateRestDmlOperation(TIcebergOperationParam operation)
throws ImpalaRuntimeException {
- if (operation.getOperation() != TIcebergOperation.INSERT) {
+ if (!SUPPORTED_REST_OPS.contains(operation.getOperation())) {
throw new ImpalaRuntimeException(
- "The Iceberg REST finalizer only supports INSERT INTO and INSERT OVERWRITE");
+ "Unsupported operation for the Iceberg REST finalizer");
}
if (!operation.isSetTable_uuid()) {
throw new ImpalaRuntimeException(
diff --git a/fe/src/test/java/org/apache/impala/analysis/AnalyzerCapabilityTest.java b/fe/src/test/java/org/apache/impala/analysis/AnalyzerCapabilityTest.java
index 6de33c8d11..d027a60ee9 100644
--- a/fe/src/test/java/org/apache/impala/analysis/AnalyzerCapabilityTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/AnalyzerCapabilityTest.java
@@ -34,7 +34,7 @@ import org.junit.Test;
public class AnalyzerCapabilityTest {
private static final String WRITE_NOT_SUPPORTED =
- "Only INSERT INTO and INSERT OVERWRITE are supported";
+ "Only INSERT INTO, INSERT OVERWRITE and UPDATE are supported";
@Test
public void testInsertCapabilityOnlyBypassesHmsAccessType() throws Exception {
@@ -48,7 +48,7 @@ public class AnalyzerCapabilityTest {
when(table.getMetaStoreTable()).thenReturn(msTable);
when(table.getFullName()).thenReturn("rest_db.rest_table");
when(table.hasProviderDerivedCapabilities()).thenReturn(true);
- when(table.supportsInsert()).thenReturn(true);
+ when(table.supportsDml()).thenReturn(true);
org.apache.iceberg.Table apiTable = mock(org.apache.iceberg.Table.class);
when(apiTable.uuid()).thenReturn(UUID.randomUUID());
when(table.getIcebergApiTable()).thenReturn(apiTable);
@@ -70,7 +70,7 @@ public class AnalyzerCapabilityTest {
FeIcebergTable table = mock(FeIcebergTable.class);
when(table.getFullName()).thenReturn("rest_db.rest_table");
when(table.hasProviderDerivedCapabilities()).thenReturn(true);
- when(table.supportsInsert()).thenReturn(false);
+ when(table.supportsDml()).thenReturn(false);
assertInsertRejected(table,
"REST catalog does not have a unique name");
@@ -82,7 +82,7 @@ public class AnalyzerCapabilityTest {
FeIcebergTable table = mock(FeIcebergTable.class);
when(table.getFullName()).thenReturn("rest_db.rest_table");
when(table.hasProviderDerivedCapabilities()).thenReturn(true);
- when(table.supportsInsert()).thenReturn(true);
+ when(table.supportsDml()).thenReturn(true);
assertInsertRejected(table,
"Iceberg table UUID is unavailable");
diff --git a/fe/src/test/java/org/apache/impala/service/IcebergRestDmlOperationTest.java b/fe/src/test/java/org/apache/impala/service/IcebergRestDmlOperationTest.java
index b6fd72d47a..45d5ee9a86 100644
--- a/fe/src/test/java/org/apache/impala/service/IcebergRestDmlOperationTest.java
+++ b/fe/src/test/java/org/apache/impala/service/IcebergRestDmlOperationTest.java
@@ -35,7 +35,8 @@ import org.junit.Test;
public class IcebergRestDmlOperationTest {
@Test
- public void testStandaloneFinalizerAcceptsOnlyInsert() throws Exception {
+ public void testStandaloneFinalizerAcceptsInsertAndUpdate()
+ throws Exception {
TIcebergOperationParam operation = new TIcebergOperationParam();
operation.setOperation(TIcebergOperation.INSERT);
operation.setIs_overwrite(false);
@@ -50,8 +51,12 @@ public class IcebergRestDmlOperationTest {
operation.setTable_uuid(UUID.randomUUID().toString());
operation.setIs_overwrite(false);
+ operation.setOperation(TIcebergOperation.UPDATE);
+ IcebergDmlFinalizer.validateRestDmlOperation(operation);
+
operation.setOperation(TIcebergOperation.DELETE);
- assertUnsupported(operation);
+ assertUnsupported(operation,
+ "Unsupported operation for the Iceberg REST finalizer");
}
@Test
@@ -83,14 +88,15 @@ public class IcebergRestDmlOperationTest {
assertFalse(params.isSetRest_catalog_name());
}
- private static void assertUnsupported(TIcebergOperationParam operation)
+ private static void assertUnsupported(
+ TIcebergOperationParam operation, String expectedMessage)
throws Exception {
try {
IcebergDmlFinalizer.validateRestDmlOperation(operation);
fail("Expected unsupported REST DML operation to be rejected");
} catch (ImpalaRuntimeException e) {
- assertTrue(e.getMessage().contains(
- "only supports INSERT INTO and INSERT OVERWRITE"));
+ assertTrue(e.getMessage(), e.getMessage().contains(
+ expectedMessage));
}
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update-partitions.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update-partitions.test
new file mode 100644
index 0000000000..0303e06a69
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update-partitions.test
@@ -0,0 +1,243 @@
+====
+---- QUERY
+# Adapted from iceberg-v3-update-partitions.test for REST catalog.
+# Tables are pre-created via the REST API; DDL tests (CREATE TABLE,
+# ALTER TABLE, partition evolution) are omitted.
+# TODO: Remove this file and run iceberg-v3-update-partitions.test against the REST
+# catalog once CREATE TABLE, CTAS and ALTER TABLE are supported for REST catalog tables.
+# Identity partition: UPDATE on empty table is no-op.
+UPDATE id_part SET i = 1;
+---- DML_RESULTS: id_part
+---- TYPES
+INT,STRING
+====
+---- QUERY
+INSERT INTO id_part VALUES(1, 'one'), (2, 'two'), (3, 'three');
+UPDATE id_part SET i = cast(i * 10 as int);
+---- DML_RESULTS: id_part
+10,'one'
+20,'two'
+30,'three'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+INSERT INTO id_part VALUES(4, 'four'), (5, 'five'), (6, 'six');
+UPDATE id_part SET i = cast(i / 2 as int), s = upper(s);
+---- DML_RESULTS: id_part
+5,'ONE'
+10,'TWO'
+15,'THREE'
+2,'FOUR'
+2,'FIVE'
+3,'SIX'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Truncate partition: UPDATE on empty table is no-op.
+UPDATE trunc_part SET i = 1;
+---- DML_RESULTS: trunc_part
+---- TYPES
+INT,STRING
+====
+---- QUERY
+INSERT INTO trunc_part VALUES(1, 'one'), (2, 'two'), (3, 'three');
+UPDATE trunc_part SET i = cast(i + 100 as int) WHERE s like 't%';
+---- DML_RESULTS: trunc_part
+1,'one'
+102,'two'
+103,'three'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+INSERT INTO trunc_part VALUES(4, 'four'), (5, 'five'), (6, 'six');
+UPDATE trunc_part SET s = concat(upper(s), s), i = cast(i + 1000 as int) WHERE i % 2 = 0;
+---- DML_RESULTS: trunc_part
+1,'one'
+1102,'TWOtwo'
+103,'three'
+1004,'FOURfour'
+5,'five'
+1006,'SIXsix'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# Multi-partition: bucket + truncate.
+# UPDATE on empty table is no-op.
+UPDATE multi_part SET i = 1;
+---- DML_RESULTS: multi_part
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+INSERT INTO multi_part VALUES(1, 'one', 1.1), (2, 'two', 2.2), (3, 'three', 3.33);
+UPDATE multi_part
+SET s = concat(s, s), f = 9.9, i = cast(i + 10 as int)
+WHERE i != (select min(i) from multi_part) and
+ i != (select max(i) from multi_part);
+---- DML_RESULTS: multi_part
+1,'one',1.1
+12,'twotwo',9.9
+3,'three', 3.33
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+INSERT INTO multi_part VALUES(4, 'four', 4.4), (5, 'five', 5.5), (6, 'six', 6.6);
+UPDATE multi_part SET i = 0, s = 'void', f = 3.14 WHERE i % 2 = 0;
+---- DML_RESULTS: multi_part
+1,'one',1.1
+0,'void',3.14
+3,'three', 3.33
+0,'void',3.14
+5,'five',5.5
+0,'void',3.14
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+# No matching rows to WHERE condition
+UPDATE multi_part SET i = 0, s = 'void', f = 3.14
+WHERE f > 1000;
+---- DML_RESULTS: multi_part
+1,'one',1.1
+0,'void',3.14
+3,'three',3.33
+0,'void',3.14
+5,'five',5.5
+0,'void',3.14
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+INSERT INTO multi_part VALUES (null, 'null',0.0);
+---- DML_RESULTS: multi_part
+1,'one',1.1
+0,'void',3.14
+3,'three',3.33
+0,'void',3.14
+5,'five',5.5
+0,'void',3.14
+NULL,'null',0.0
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+UPDATE multi_part SET i = 111, s = 'fox', f = 1.1 where i is null;
+---- DML_RESULTS: multi_part
+1,'one',1.1
+0,'void',3.14
+3,'three',3.33
+0,'void',3.14
+5,'five',5.5
+0,'void',3.14
+111,'fox',1.1
+---- TYPES
+INT,STRING,DOUBLE
+====
+---- QUERY
+# Date partition: year transform.
+insert into date_year_part values
+ (1, '1968-01-01'), (2, '1969-12-31'), (3, '1970-01-01'), (4, '2023-11-15');
+UPDATE date_year_part SET d = years_add(d, 10), i = cast(i + 10 as int);
+---- DML_RESULTS: date_year_part
+11,1978-01-01
+12,1979-12-31
+13,1980-01-01
+14,2033-11-15
+---- TYPES
+INT, DATE
+====
+---- QUERY
+# Date partition: month transform.
+insert into date_month_part values
+ (1, '1968-01-01'), (2, '1969-12-31'), (3, '1970-01-01'), (4, '2023-11-15');
+UPDATE date_month_part SET d = years_add(d, 10), i = cast(i + 10 as int);
+---- DML_RESULTS: date_month_part
+11,1978-01-01
+12,1979-12-31
+13,1980-01-01
+14,2033-11-15
+---- TYPES
+INT, DATE
+====
+---- QUERY
+# Date partition: day transform.
+insert into date_day_part values
+ (1, '1968-01-01'), (2, '1969-12-31'), (3, '1970-01-01'), (4, '2023-11-15');
+UPDATE date_day_part SET d = years_add(d, 10), i = cast(i + 10 as int);
+---- DML_RESULTS: date_day_part
+11,1978-01-01
+12,1979-12-31
+13,1980-01-01
+14,2033-11-15
+---- TYPES
+INT, DATE
+====
+---- QUERY
+# Timestamp partition: year transform.
+insert into ts_year_part values (1, '1968-01-01'),
+ (2, '1969-12-31'), (3, '1970-01-01'), (4, '2023-11-15');
+UPDATE ts_year_part SET ts = years_add(ts, -10), i = cast(i + 100 as int);
+---- DML_RESULTS: ts_year_part
+101,1958-01-01 00:00:00
+102,1959-12-31 00:00:00
+103,1960-01-01 00:00:00
+104,2013-11-15 00:00:00
+---- TYPES
+INT, TIMESTAMP
+====
+---- QUERY
+# Timestamp partition: month transform.
+insert into ts_month_part values (1, '1968-01-01 01:02:03'),
+ (2, '1969-12-31 23:59:00'), (3, '1970-01-01 00:00:00'),
+ (4, '2023-11-15 15:31:00');
+UPDATE ts_month_part SET ts = years_add(ts, -10), i = cast(i + 100 as int);
+---- DML_RESULTS: ts_month_part
+101,1958-01-01 01:02:03
+102,1959-12-31 23:59:00
+103,1960-01-01 00:00:00
+104,2013-11-15 15:31:00
+---- TYPES
+INT, TIMESTAMP
+====
+---- QUERY
+# Timestamp partition: day transform.
+insert into ts_day_part values (1, '1968-01-01 01:02:03'),
+ (2, '1969-12-31 23:59:00'), (3, '1970-01-01 00:00:00'),
+ (4, '2023-11-15 15:31:00');
+UPDATE ts_day_part SET ts = years_add(ts, -10), i = cast(i + 100 as int);
+---- DML_RESULTS: ts_day_part
+101,1958-01-01 01:02:03
+102,1959-12-31 23:59:00
+103,1960-01-01 00:00:00
+104,2013-11-15 15:31:00
+---- TYPES
+INT, TIMESTAMP
+====
+---- QUERY
+# Timestamp partition: hour transform.
+insert into ts_hour_part values (1, '1968-01-01 01:02:03'),
+ (2, '1969-12-31 23:59:00'), (3, '1970-01-01 00:00:00'),
+ (4, '2023-11-15 15:31:00');
+UPDATE ts_hour_part SET ts = years_add(ts, -10), i = cast(i + 100 as int);
+---- DML_RESULTS: ts_hour_part
+101,1958-01-01 01:02:03
+102,1959-12-31 23:59:00
+103,1960-01-01 00:00:00
+104,2013-11-15 15:31:00
+---- TYPES
+INT, TIMESTAMP
+====
+---- QUERY
+# Numeric truncate with multiple partition transforms.
+insert into numeric_truncate values (1, 12, 1222, 135, 20.75);
+UPDATE numeric_truncate SET dec_10_2 = 75.20, dec_8_0 = 531, bigint_col = 2111, int_col = 21, id = 11;
+---- DML_RESULTS: numeric_truncate
+11,21,2111,531,75.20
+---- TYPES
+INT,INT,BIGINT,DECIMAL,DECIMAL
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update.test
new file mode 100644
index 0000000000..311499d7c6
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-rest-update.test
@@ -0,0 +1,427 @@
+====
+---- QUERY
+# Adapted from iceberg-v3-update.test for REST catalog.
+# Tables are pre-created via the REST API; DDL tests (CREATE TABLE,
+# ALTER TABLE, DELETE, CREATE VIEW, UDF) are omitted.
+# TODO: Remove this file and run iceberg-v3-update.test against the REST catalog
+# once CREATE TABLE, CTAS and ALTER TABLE are supported for REST catalog tables.
+# Basic UPDATE on unpartitioned V3 table via REST catalog.
+INSERT INTO update_basic VALUES (1, 'one'), (2, 'two'), (3, 'three');
+UPDATE update_basic SET s = 'updated' WHERE i = 2;
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+SELECT * FROM update_basic ORDER BY i;
+---- RESULTS
+1,'one'
+2,'updated'
+3,'three'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# _row_id is preserved, _last_updated_sequence_number is updated.
+SELECT _row_id, _last_updated_sequence_number, '|', * FROM update_basic ORDER BY i;
+---- RESULTS
+0,1,'|',1,'one'
+1,2,'|',2,'updated'
+2,1,'|',3,'three'
+---- TYPES
+BIGINT,BIGINT,STRING,INT,STRING
+====
+---- QUERY
+# The delete side of UPDATE creates a DV (Puffin) file.
+SELECT content, file_format, record_count, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.update_basic.`files`
+ORDER BY content, file_format;
+---- RESULTS
+0,'PARQUET',1,false,false
+0,'PARQUET',3,false,false
+1,'PUFFIN',1,true,true
+---- TYPES
+INT,STRING,BIGINT,BOOLEAN,BOOLEAN
+====
+---- QUERY
+# UPDATE multiple rows at once.
+INSERT INTO update_multi VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd'), (5, 'e');
+UPDATE update_multi SET s = upper(s) WHERE i >= 3;
+---- RUNTIME_PROFILE
+NumModifiedRows: 3
+NumDeletedRows: 3
+====
+---- QUERY
+SELECT * FROM update_multi ORDER BY i;
+---- RESULTS
+1,'a'
+2,'b'
+3,'C'
+4,'D'
+5,'E'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+# _row_id is preserved, _last_updated_sequence_number is updated.
+SELECT _row_id, _last_updated_sequence_number, '|', * FROM update_multi ORDER BY i;
+---- RESULTS
+0,1,'|',1,'a'
+1,1,'|',2,'b'
+2,2,'|',3,'C'
+3,2,'|',4,'D'
+4,2,'|',5,'E'
+---- TYPES
+BIGINT,BIGINT,STRING,INT,STRING
+====
+---- QUERY
+# No WHERE clause, but only update a subset of rows.
+UPDATE update_multi SET s = lower(s);
+---- RUNTIME_PROFILE
+NumModifiedRows: 3
+NumDeletedRows: 3
+====
+---- QUERY
+SELECT * FROM update_multi ORDER BY i;
+---- RESULTS
+1,'a'
+2,'b'
+3,'c'
+4,'d'
+5,'e'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+SELECT content, file_format, record_count, referenced_data_file IS NOT NULL, content_offset IS NOT NULL
+FROM $DATABASE.update_multi.`files`
+ORDER BY content, file_format, record_count;
+---- RESULTS
+0,'PARQUET',3,false,false
+0,'PARQUET',3,false,false
+0,'PARQUET',5,false,false
+1,'PUFFIN',3,true,true
+1,'PUFFIN',3,true,true
+---- TYPES
+INT,STRING,BIGINT,BOOLEAN,BOOLEAN
+====
+---- QUERY
+UPDATE update_multi SET s = 'x';
+---- RUNTIME_PROFILE
+NumModifiedRows: 5
+NumDeletedRows: 5
+====
+---- QUERY
+SELECT * FROM update_multi ORDER BY i;
+---- RESULTS
+1,'x'
+2,'x'
+3,'x'
+4,'x'
+5,'x'
+---- TYPES
+INT,STRING
+====
+---- QUERY
+update single_col set i = 4;
+---- DML_RESULTS: single_col
+---- TYPES
+INT
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 0
+====
+---- QUERY
+insert into single_col values (1), (2), (3);
+update single_col set i = cast(i + 1 as int);
+---- DML_RESULTS: single_col
+2
+3
+4
+---- TYPES
+INT
+---- RUNTIME_PROFILE
+NumModifiedRows: 3
+NumDeletedRows: 3
+====
+---- QUERY
+update single_col set i = 1 where i = 2
+---- DML_RESULTS: single_col
+1
+3
+4
+---- TYPES
+INT
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+insert into ice_alltypes values (false, 0, 0, 0, 0, 0, 0, '2000-01-01', '2000-01-01 00:00:00',
+ 'zero', cast('zerob' as binary));
+---- DML_RESULTS: ice_alltypes
+false,0,0,0,0,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 0
+====
+---- QUERY
+update ice_alltypes set bool_col = true;
+---- DML_RESULTS: ice_alltypes
+true,0,0,0,0,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set int_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,0,0,0,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set bigint_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,1,0,0,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set float_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,0,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set double_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,0,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set dec_10_0_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,0.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set dec_8_2_col = 1;
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,1.00,2000-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set date_col = '2001-01-01';
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,1.00,2001-01-01,2000-01-01 00:00:00,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set timestamp_col = '2001-01-01 01:01:01';
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'zero','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set string_col = 'one';
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'one','zerob'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set binary_col = cast('oneb' as binary);
+---- DML_RESULTS: ice_alltypes
+true,1,1,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'one','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set bigint_col = 33, int_col = 30, string_col = 'three';
+---- DML_RESULTS: ice_alltypes
+true,30,33,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'three','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# The row already has the desired values, no need to update. (IMPALA-12588)
+update ice_alltypes set bigint_col = 33, int_col = 30, string_col = 'three';
+---- DML_RESULTS: ice_alltypes
+true,30,33,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'three','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 0
+====
+---- QUERY
+# Only one field changes.
+update ice_alltypes set bigint_col = 33, int_col = 3, string_col = 'three';
+---- DML_RESULTS: ice_alltypes
+true,3,33,1,1,1,1.00,2001-01-01,2001-01-01 01:01:01,'three','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+update ice_alltypes set dec_10_0_col = 23, dec_8_2_col = cast(123.123 as decimal(8, 2));
+---- DML_RESULTS: ice_alltypes
+true,3,33,1,1,23,123.12,2001-01-01,2001-01-01 01:01:01,'three','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# UPDATE table based on a reference table.
+insert into ice_alltypes values (false, 0, 0, 0, 0, 0, 0, '2000-01-01', '2000-01-01 00:00:00',
+ 'zero', cast('zerob' as binary));
+insert into ref_table values (0, 111, 'IMPALA', '2023-11-07'), (3, 222, 'ICEBERG', '2023-11-08');
+update ice_alltypes set bigint_col=bi, string_col=s, date_col=d from ice_alltypes, ref_table where int_col = i;
+---- DML_RESULTS: ice_alltypes
+false,0,111,0,0,0,0.00,2023-11-07,2000-01-01 00:00:00,'IMPALA','zerob'
+true,3,222,1,1,23,123.12,2023-11-08,2001-01-01 01:01:01,'ICEBERG','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 2
+NumDeletedRows: 2
+====
+---- QUERY
+# If the JOIN in UPDATE has multiple matches Impala should raise an error.
+# Changing bigint_col to actually have rows to update.
+insert into ref_table values (0, 1111, 'IIMMPPAALLAA', '2023-12-01');
+update ice_alltypes set bigint_col=bi+1, string_col=s, date_col=d from ice_alltypes, ref_table where int_col = i;
+---- CATCH
+Duplicated row in DELETE sink.
+====
+---- QUERY
+# Check that NULL values and expressions evaluating to NULL in assignments are handled correctly.
+update ice_alltypes set dec_10_0_col = NULL, dec_8_2_col = cast(123.123 + NULL as decimal(8, 2));
+---- DML_RESULTS: ice_alltypes
+false,0,111,0,0,NULL,NULL,2023-11-07,2000-01-01 00:00:00,'IMPALA','zerob'
+true,3,222,1,1,NULL,NULL,2023-11-08,2001-01-01 01:01:01,'ICEBERG','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 2
+NumDeletedRows: 2
+====
+---- QUERY
+# Check that NULL assignments and WHERE condition filter rows correctly.
+update ice_alltypes set dec_10_0_col = cast(123.123 as decimal(10, 0)), dec_8_2_col = NULL where int_col < 2;
+---- DML_RESULTS: ice_alltypes
+false,0,111,0,0,123,NULL,2023-11-07,2000-01-01 00:00:00,'IMPALA','zerob'
+true,3,222,1,1,NULL,NULL,2023-11-08,2001-01-01 01:01:01,'ICEBERG','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# Swap two columns
+update ice_alltypes set dec_10_0_col = cast(dec_8_2_col as decimal(10, 0)), dec_8_2_col = cast(dec_10_0_col as decimal(8, 2));
+---- DML_RESULTS: ice_alltypes
+false,0,111,0,0,NULL,123.00,2023-11-07,2000-01-01 00:00:00,'IMPALA','zerob'
+true,3,222,1,1,NULL,NULL,2023-11-08,2001-01-01 01:01:01,'ICEBERG','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# Check col_a = col_b + col_c
+update ice_alltypes set dec_10_0_col = cast(bigint_col + dec_8_2_col as decimal(10, 0));
+---- DML_RESULTS: ice_alltypes
+false,0,111,0,0,234,123.00,2023-11-07,2000-01-01 00:00:00,'IMPALA','zerob'
+true,3,222,1,1,NULL,NULL,2023-11-08,2001-01-01 01:01:01,'ICEBERG','oneb'
+---- TYPES
+BOOLEAN,INT,BIGINT,FLOAT,DOUBLE,DECIMAL,DECIMAL,DATE,TIMESTAMP,STRING,BINARY
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# Test if changing SKIP_UNNEEDED_UPDATES_COL_LIMIT query option takes effect: the row
+# is updated even though the table content remained the same.
+UPDATE update_basic SET i = cast(i * 10 as int), s = 'same';
+---- RUNTIME_PROFILE
+NumModifiedRows: 3
+NumDeletedRows: 3
+====
+---- QUERY
+SET SKIP_UNNEEDED_UPDATES_COL_LIMIT = 1;
+UPDATE update_basic SET i = 10, s = 'same' WHERE i = 10;
+---- DML_RESULTS: update_basic
+10,'same'
+20,'same'
+30,'same'
+---- TYPES
+INT,STRING
+---- RUNTIME_PROFILE
+NumModifiedRows: 1
+NumDeletedRows: 1
+====
+---- QUERY
+# Setting SKIP_UNNEEDED_UPDATES_COL_LIMIT back to default. No rows should be updated.
+SET SKIP_UNNEEDED_UPDATES_COL_LIMIT = 10;
+UPDATE update_basic SET i = 10, s = 'same' WHERE i = 10;
+---- DML_RESULTS: update_basic
+10,'same'
+20,'same'
+30,'same'
+---- TYPES
+INT,STRING
+---- RUNTIME_PROFILE
+NumModifiedRows: 0
+NumDeletedRows: 0
+====
diff --git a/tests/common/iceberg_rest_test_tables.py b/tests/common/iceberg_rest_test_tables.py
new file mode 100644
index 0000000000..1c2607c2d0
--- /dev/null
+++ b/tests/common/iceberg_rest_test_tables.py
@@ -0,0 +1,131 @@
+# 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.
+
+"""Definitions of the Iceberg REST catalog tables used by the REST DML tests. CREATE
+TABLE is not supported for REST catalog tables, so IcebergRestCatalogTests creates the
+tables declared in RestServerProperties through the REST API.
+
+A definition is a dict with 'fields', and optionally 'partition' (partition spec
+fields) and 'properties' (table properties, including 'format-version')."""
+
+
+def _rest_col(field_id, name, col_type):
+ return {'id': field_id, 'name': name, 'required': False, 'type': col_type}
+
+
+def _rest_part(source_id, field_id, name, transform):
+ return {'source-id': source_id, 'field-id': field_id, 'name': name,
+ 'transform': transform}
+
+
+def _v3(tables):
+ return dict((name, dict(table, properties=dict(
+ table.get('properties', {}), **{'format-version': '3'})))
+ for name, table in tables.items())
+
+
+def create_table_request(table_name, table):
+ """Returns the REST API request that creates 'table_name' from its definition."""
+ request = {
+ 'name': table_name,
+ 'schema': {'type': 'struct', 'schema-id': 0, 'fields': table['fields']},
+ 'properties': table.get('properties', {}),
+ }
+ if 'partition' in table:
+ request['partition-spec'] = {'spec-id': 0, 'fields': table['partition']}
+ return request
+
+
+# The 'test_table' and 'partitioned_test_table' of RestServerProperties.
+INSERT_TEST_TABLE = {
+ 'fields': [_rest_col(1, 'i', 'int')],
+ 'properties': {'format-version': '2'}}
+PARTITIONED_INSERT_TEST_TABLE = {
+ 'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 'p', 'int')],
+ 'partition': [_rest_part(2, 1000, 'p', 'identity')],
+ 'properties': {'format-version': '2'}}
+
+
+# Tables of iceberg-rest-update.test, adapted from iceberg-v3-update.test.
+UPDATE_TEST_TABLES = _v3({
+ 'update_basic': {'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 's', 'string')]},
+ 'update_multi': {'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 's', 'string')]},
+ 'single_col': {'fields': [_rest_col(1, 'i', 'int')]},
+ 'ice_alltypes': {'fields': [
+ _rest_col(1, 'bool_col', 'boolean'), _rest_col(2, 'int_col', 'int'),
+ _rest_col(3, 'bigint_col', 'long'), _rest_col(4, 'float_col', 'float'),
+ _rest_col(5, 'double_col', 'double'),
+ _rest_col(6, 'dec_10_0_col', 'decimal(10, 0)'),
+ _rest_col(7, 'dec_8_2_col', 'decimal(8, 2)'),
+ _rest_col(8, 'date_col', 'date'), _rest_col(9, 'timestamp_col', 'timestamp'),
+ _rest_col(10, 'string_col', 'string'), _rest_col(11, 'binary_col', 'binary')]},
+ 'ref_table': {'fields': [
+ _rest_col(1, 'i', 'int'), _rest_col(2, 'bi', 'long'),
+ _rest_col(3, 's', 'string'), _rest_col(4, 'd', 'date')]},
+})
+
+
+def _int_and(name, col_type):
+ return [_rest_col(1, 'i', 'int'), _rest_col(2, name, col_type)]
+
+
+# Mirrors the CREATE TABLE statements of iceberg-v3-update-partitions.test.
+UPDATE_PARTITIONS_TEST_TABLES = _v3({
+ 'id_part': {
+ 'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 's', 'string')],
+ 'partition': [_rest_part(1, 1000, 'i', 'identity')]},
+ 'trunc_part': {
+ 'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 's', 'string')],
+ 'partition': [_rest_part(2, 1000, 's_trunc', 'truncate[1]')]},
+ 'multi_part': {
+ 'fields': [_rest_col(1, 'i', 'int'), _rest_col(2, 's', 'string'),
+ _rest_col(3, 'f', 'double')],
+ 'partition': [_rest_part(1, 1000, 'i_bucket', 'bucket[7]'),
+ _rest_part(2, 1001, 's_trunc', 'truncate[1]')],
+ 'properties': {'sort.columns': 'f,i', 'sort.order': 'LEXICAL'}},
+ 'date_year_part': {
+ 'fields': _int_and('d', 'date'),
+ 'partition': [_rest_part(2, 1000, 'd_year', 'year')]},
+ 'date_month_part': {
+ 'fields': _int_and('d', 'date'),
+ 'partition': [_rest_part(2, 1000, 'd_month', 'month')]},
+ 'date_day_part': {
+ 'fields': _int_and('d', 'date'),
+ 'partition': [_rest_part(2, 1000, 'd_day', 'day')]},
+ 'ts_year_part': {
+ 'fields': _int_and('ts', 'timestamp'),
+ 'partition': [_rest_part(2, 1000, 'ts_year', 'year')]},
+ 'ts_month_part': {
+ 'fields': _int_and('ts', 'timestamp'),
+ 'partition': [_rest_part(2, 1000, 'ts_month', 'month')]},
+ 'ts_day_part': {
+ 'fields': _int_and('ts', 'timestamp'),
+ 'partition': [_rest_part(2, 1000, 'ts_day', 'day')]},
+ 'ts_hour_part': {
+ 'fields': _int_and('ts', 'timestamp'),
+ 'partition': [_rest_part(2, 1000, 'ts_hour', 'hour')]},
+ 'numeric_truncate': {
+ 'fields': [_rest_col(1, 'id', 'int'), _rest_col(2, 'int_col', 'int'),
+ _rest_col(3, 'bigint_col', 'long'),
+ _rest_col(4, 'dec_8_0', 'decimal(8, 0)'),
+ _rest_col(5, 'dec_10_2', 'decimal(10, 2)')],
+ 'partition': [_rest_part(2, 1000, 'int_col_trunc', 'truncate[10]'),
+ _rest_part(3, 1001, 'bigint_col_trunc', 'truncate[1000]'),
+ _rest_part(1, 1002, 'id_null', 'void'),
+ _rest_part(4, 1003, 'dec_8_0_trunc', 'truncate[20]'),
+ _rest_part(5, 1004, 'dec_10_2_trunc', 'truncate[50]')]},
+})
diff --git a/tests/custom_cluster/test_iceberg_rest_catalog.py b/tests/custom_cluster/test_iceberg_rest_catalog.py
index fbf4affaac..cdd66067af 100644
--- a/tests/custom_cluster/test_iceberg_rest_catalog.py
+++ b/tests/custom_cluster/test_iceberg_rest_catalog.py
@@ -24,6 +24,9 @@ import requests
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite, HIVE_CONF_DIR
from tests.common.iceberg_rest_server import IcebergRestServer
+from tests.common.iceberg_rest_test_tables import (
+ INSERT_TEST_TABLE, PARTITIONED_INSERT_TEST_TABLE, UPDATE_PARTITIONS_TEST_TABLES,
+ UPDATE_TEST_TABLES, create_table_request)
from tests.common.impala_connection import IMPALA_CONNECTION_EXCEPTION
from tests.util.filesystem_utils import get_fs_path
from tests.util.hdfs_util import HadoopFsCommandLineClient
@@ -51,14 +54,17 @@ SECONDARY_WRITE_TEST_TABLE = "impala_rest_dml_test_secondary.insert_test"
WRITE_TEST_CATALOG_LOCATION = IcebergRestServer.DEFAULT_CATALOG_LOCATION
SECONDARY_WRITE_TEST_CATALOG_LOCATION = \
'/test-warehouse/iceberg_test/secondary_hadoop_catalog'
+UPDATE_TEST_NS = "impala_rest_update_test"
def RestServerProperties(*server_configs):
"""
Annotation to specify configurations for multiple REST servers to be started.
Each argument is a dictionary with optional 'port', 'catalog_location',
- 'test_table' and 'partitioned_test_table' keys. The test tables are created through
- the REST API, see _create_rest_test_table().
+ 'test_table', 'partitioned_test_table' and 'test_tables' keys. 'test_tables' is a
+ (namespace, {table name: definition}) pair, see iceberg_rest_test_tables. The test
+ tables are created through the REST API before the test and dropped after it, see
+ _create_rest_test_table().
Example:
@RestServerProperties({'port': 9085},
{'port': 9086, 'catalog_location': '/tmp/cat2', 'test_table': 'db.tbl'})
@@ -77,8 +83,8 @@ class IcebergRestCatalogTests(CustomClusterTestSuite):
raise Exception("Cannot specify HIVE_CONF_DIR because the tests of this class are "
"running without Hive.")
self.servers = []
+ # (server, table name, table definition) of the tables to create.
self.rest_test_tables = []
- self.partitioned_rest_test_tables = set()
server_configs = getattr(method, 'rest_server_configs', None)
@@ -94,17 +100,21 @@ class IcebergRestCatalogTests(CustomClusterTestSuite):
server = IcebergRestServer(port, catalog_location)
self.servers.append(server)
if test_table:
- self.rest_test_tables.append((server, test_table))
+ self.rest_test_tables.append((server, test_table, INSERT_TEST_TABLE))
if partitioned_test_table:
- self.rest_test_tables.append((server, partitioned_test_table))
- self.partitioned_rest_test_tables.add(partitioned_test_table)
+ self.rest_test_tables.append(
+ (server, partitioned_test_table, PARTITIONED_INSERT_TEST_TABLE))
+ if 'test_tables' in config:
+ namespace, tables = config['test_tables']
+ for table, definition in tables.items():
+ self.rest_test_tables.append(
+ (server, "{}.{}".format(namespace, table), definition))
try:
for server in self.servers:
server.start_rest_server(300)
- for server, table_name in self.rest_test_tables:
- self._create_rest_test_table(server, table_name,
- partitioned=table_name in self.partitioned_rest_test_tables)
+ for server, table_name, definition in self.rest_test_tables:
+ self._create_rest_test_table(server, table_name, definition=definition)
super(IcebergRestCatalogTests, self).setup_method(method)
# At this point we can create the Impala clients that we will need.
self.create_impala_clients()
@@ -124,7 +134,7 @@ class IcebergRestCatalogTests(CustomClusterTestSuite):
def _stop_rest_servers(self, timeout_s=60):
first_error = None
- for server, table_name in getattr(self, 'rest_test_tables', []):
+ for server, table_name, _ in getattr(self, 'rest_test_tables', []):
try:
self._drop_rest_test_table(server, table_name)
except Exception as e:
@@ -153,9 +163,13 @@ class IcebergRestCatalogTests(CustomClusterTestSuite):
response.raise_for_status()
@classmethod
- def _create_rest_test_table(cls, server, table_name, partitioned=False):
- """Creates 'table_name' with an INT column 'i'. A partitioned table also gets an INT
- column 'p' and an identity partition spec on it."""
+ def _create_rest_test_table(cls, server, table_name, partitioned=False,
+ definition=None):
+ """Creates 'table_name' from 'definition', see iceberg_rest_test_tables. Without a
+ definition, the table gets an INT column 'i', and a partitioned table also gets an
+ INT column 'p' and an identity partition spec on it."""
+ if definition is None:
+ definition = PARTITIONED_INSERT_TEST_TABLE if partitioned else INSERT_TEST_TABLE
namespace, table = table_name.split('.', 1)
cls._drop_rest_test_table(server, table_name)
namespace_response = requests.post(
@@ -164,27 +178,10 @@ class IcebergRestCatalogTests(CustomClusterTestSuite):
if namespace_response.status_code != 409:
namespace_response.raise_for_status()
- fields = [{'id': 1, 'name': 'i', 'required': False, 'type': 'int'}]
- create_request = {
- 'name': table,
- 'schema': {
- 'type': 'struct',
- 'schema-id': 0,
- 'fields': fields,
- },
- 'properties': {'format-version': '2'},
- }
- if partitioned:
- fields.append({'id': 2, 'name': 'p', 'required': False, 'type': 'int'})
- create_request['partition-spec'] = {
- 'spec-id': 0,
- 'fields': [{'source-id': 2, 'field-id': 1000, 'name': 'p',
- 'transform': 'identity'}],
- }
create_response = requests.post(
"http://localhost:{}/v1/namespaces/{}/tables".format(
server.port, quote(namespace, safe='')),
- json=create_request, timeout=30)
+ json=create_table_request(table, definition), timeout=30)
create_response.raise_for_status()
@@ -306,7 +303,6 @@ class TestIcebergRestCatalogDmlStandalone(IcebergRestCatalogTests):
unsupported_statements = [
"truncate table {}".format(table_name),
"delete from {} where i = 1".format(table_name),
- "update {} set i = 2 where i = 1".format(table_name),
"merge into {0} target using (select 1 i) source "
"on target.i = source.i when matched then update set i = 2".format(
table_name),
@@ -314,8 +310,76 @@ class TestIcebergRestCatalogDmlStandalone(IcebergRestCatalogTests):
]
for statement in unsupported_statements:
error = self.execute_query_expect_failure(self.client, statement)
- assert ("AnalysisException: Only INSERT INTO and INSERT OVERWRITE are supported"
- in str(error))
+ assert ("AnalysisException: Only INSERT INTO, INSERT OVERWRITE and UPDATE "
+ "are supported") in str(error)
+
+ @RestServerProperties(
+ {'port': 9084, 'catalog_location': WRITE_TEST_CATALOG_LOCATION,
+ 'test_tables': (UPDATE_TEST_NS, UPDATE_TEST_TABLES)})
+ @CustomClusterTestSuite.with_args(
+ impalad_args=REST_STANDALONE_IMPALAD_ARGS,
+ start_args=NO_CATALOGD_STARTARGS,
+ cluster_size=1)
+ @pytest.mark.execute_serially
+ def test_rest_catalog_update(self, vector):
+ """Test UPDATE on Iceberg REST catalog tables (format-version 3)."""
+ ns = UPDATE_TEST_NS
+ self.run_test_case('QueryTest/iceberg-rest-update', vector, use_db=ns)
+
+ table_name = "{}.update_basic".format(ns)
+ data_path = get_fs_path(
+ "/test-warehouse/iceberg_test/hadoop_catalog/"
+ "{}/update_basic/data".format(ns))
+ filesystem_client = HadoopFsCommandLineClient()
+
+ # State left behind by the last section of iceberg-rest-update.test.
+ expected_rows = ['10\tsame', '20\tsame', '30\tsame']
+ committed_files = set(filesystem_client.ls(data_path))
+ unknown_state = {'debug_action':
+ 'ICEBERG_COMMIT:EXCEPTION@CommitStateUnknownException@'
+ 'simulated REST commit state unknown'}
+ error = self.execute_query_expect_failure(
+ self.client,
+ "update {} set s = 'fail' where i = 10".format(table_name),
+ query_options=unknown_state)
+ assert "simulated REST commit state unknown" in str(error)
+
+ result = self.execute_query_expect_success(
+ self.client, "select * from {} order by i".format(table_name))
+ assert result.data == expected_rows
+ # Files are kept when the commit state is unknown.
+ orphan_files = set(filesystem_client.ls(data_path)) - committed_files
+ assert len(orphan_files) > 0
+ for orphan_file in orphan_files:
+ assert filesystem_client.delete_file_dir(data_path + "/" + orphan_file)
+
+ commit_failure = {'debug_action':
+ 'ICEBERG_COMMIT:EXCEPTION@CommitFailedException@'
+ 'simulated REST commit failure'}
+ error = self.execute_query_expect_failure(
+ self.client,
+ "update {} set s = 'fail' where i = 10".format(table_name),
+ query_options=commit_failure)
+ assert "simulated REST commit failure" in str(error)
+
+ result = self.execute_query_expect_success(
+ self.client, "select * from {} order by i".format(table_name))
+ assert result.data == expected_rows
+ # Files written by a failed commit are cleaned up.
+ assert set(filesystem_client.ls(data_path)) == committed_files
+
+ @RestServerProperties(
+ {'port': 9084, 'catalog_location': WRITE_TEST_CATALOG_LOCATION,
+ 'test_tables': (UPDATE_TEST_NS, UPDATE_PARTITIONS_TEST_TABLES)})
+ @CustomClusterTestSuite.with_args(
+ impalad_args=REST_STANDALONE_IMPALAD_ARGS,
+ start_args=NO_CATALOGD_STARTARGS,
+ cluster_size=1)
+ @pytest.mark.execute_serially
+ def test_rest_catalog_update_partitions(self, vector):
+ """Test UPDATE on partitioned Iceberg REST catalog tables (format-version 3)."""
+ self.run_test_case(
+ 'QueryTest/iceberg-rest-update-partitions', vector, use_db=UPDATE_TEST_NS)
@RestServerProperties(
{'port': 9084, 'catalog_location': WRITE_TEST_CATALOG_LOCATION,
(impala) 01/02: IMPALA-15175: Support HBO for AnalyticEvalNode cardinality
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 0207eaa0a2779b51a7c3878d455c10b92e9eb5c6
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Mon Jul 13 17:12:36 2026 +0800
IMPALA-15175: Support HBO for AnalyticEvalNode cardinality
AnalyticEvalNode appends analytic results to the input tuples as new
columns. Though they won't change the cardinality, the analytic results
could be used in downstream group-by keys, join keys, or predicates,
which matters in cardinalities of downstream nodes. This patch adds
AnalyticEvalNode in the HBO key strings so its parent nodes can still be
tracked in HBO. Note that cardinality of AnalyticEvalNode is still
untracked since it can be simply deduced from its input cardinality.
The HBO key string of an AnalyticEvalNode consists of
- Analytic functions
- PartitionBy Exprs
- OrderBy Exprs
- Window
- The child's HBO key string
Exprs are canonicalized with operand indexes if the child node has
multiple output tuples, e.g., JoinNode. PartitionBy Exprs are sorted
since the order doesn't matter. OrderBy Exprs are added using the
original order.
Testing
- Added FE tests and e2e tests
Assisted-by: Opus 4.8 (Claude Code)
Change-Id: I8d87ab472db9273622f48fa62bb1d8c12e04e86b
Reviewed-on: http://gerrit.cloudera.org:8080/24608
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/analysis/AnalyticExpr.java | 9 +-
.../apache/impala/planner/AnalyticEvalNode.java | 68 ++++++
.../apache/impala/planner/HboKeyStringTest.java | 217 +++++++++++++++++++
.../queries/QueryTest/hbo-analytic-agg.test | 67 ++++++
.../queries/QueryTest/hbo-analytic-join.test | 229 +++++++++++++++++++++
tests/query_test/test_hbo.py | 28 +++
6 files changed, 617 insertions(+), 1 deletion(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
index 851f0bba39..2318ad1664 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
@@ -154,7 +154,14 @@ public class AnalyticExpr extends Expr {
@Override
public String toSqlImpl(ToSqlOptions options) {
- if (sqlString_ != null) return sqlString_;
+ if (options.showForHbo()) {
+ // HBO keys must render fresh so operand qualifiers applied to child SlotRefs are
+ // picked up. syncWithChildren re-derives the mirror fields from children_ and is
+ // semantics-preserving (same convention as resetAnalysisState/substituteImpl).
+ syncWithChildren();
+ } else if (sqlString_ != null) {
+ return sqlString_;
+ }
StringBuilder sb = new StringBuilder();
sb.append(fnCall_.toSql(options)).append(" OVER (");
boolean needsSpace = false;
diff --git a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
index 2ab23a6433..e0fb2872d9 100644
--- a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
@@ -18,7 +18,9 @@
package org.apache.impala.planner;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import org.apache.impala.analysis.AnalyticExpr;
import org.apache.impala.analysis.AnalyticWindow;
@@ -42,7 +44,9 @@ import org.apache.impala.catalog.Function;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TAnalyticNode;
import org.apache.impala.thrift.TExplainLevel;
+import org.apache.impala.thrift.THboStatsType;
import org.apache.impala.thrift.TPlanNode;
+import org.apache.impala.thrift.TPlanNodeRun;
import org.apache.impala.thrift.TPlanNodeType;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.ExprUtil;
@@ -132,6 +136,7 @@ public class AnalyticEvalNode extends PlanNode {
@Override
public void init(Analyzer analyzer) {
Preconditions.checkState(conjuncts_.isEmpty());
+ computeHboOperandQualifierMap(analyzer);
computeMemLayout(analyzer);
intermediateTupleDesc_.computeMemLayout();
@@ -309,6 +314,69 @@ public class AnalyticEvalNode extends PlanNode {
}
}
+ @Override
+ public boolean isOperandTransparent() { return true; }
+
+ @Override
+ public void appendScanInputStats(TPlanNodeRun execStats) {
+ getChild(0).appendScanInputStats(execStats);
+ }
+
+ /**
+ * Generates an HBO key string for this analytic node. Returns null if the child node
+ * doesn't support HBO.
+ */
+ @Override
+ public String generateHboKeyString(THboStatsType statsType,
+ CanonicalizationStrategy strategy) {
+ Preconditions.checkState(children_.size() == 1);
+ Preconditions.checkState(conjuncts_.isEmpty());
+ String childKey = getChild(0).generateHboKeyString(statsType, strategy);
+ if (childKey == null) {
+ LOG.debug("Not tracking {} in HBO since the child doesn't support HBO: {}",
+ getDisplayLabel(), getChild(0).getDisplayLabel());
+ return null;
+ }
+
+ StringBuilder sb = startHboKeyString(statsType, ":AnalyticEvalNode:");
+
+ // The analytic function calls, partition-by and order-by exprs all reference the
+ // input (the preceding sort output). When that input has multiple operands (a join
+ // below the analytic), their columns are qualified with the join's canonical operand
+ // index. A single-operand input (e.g. a scan) needs no qualification.
+ Map<TupleId, String> operandIdx = getChild(0).getHboOperandQualifierMap();
+
+ // Partition-by and function calls are order-independent and canonicalized into sorted
+ // sets.
+ List<String> fnStrs =
+ ExprCanonicalizer.canonicalizeExprs(analyticFnCalls_, strategy, operandIdx);
+ sb.append("|Fns:[").append(String.join(",", fnStrs)).append("]");
+ List<String> partStrs =
+ ExprCanonicalizer.canonicalizeExprs(partitionExprs_, strategy, operandIdx);
+ sb.append("|PartExprs:[").append(String.join(",", partStrs)).append("]");
+
+ // Order-by sequence is significant for ranking, so keep it as-is (do not sort). Each
+ // expr is qualified (via a singleton list) with the same operand index as above.
+ List<String> orderStrs = new ArrayList<>(orderByElements_.size());
+ for (OrderByElement e : orderByElements_) {
+ String exprStr = ExprCanonicalizer.canonicalizeExprs(
+ Collections.singletonList(e.getExpr()), strategy, operandIdx).get(0);
+ StringBuilder ob = new StringBuilder(exprStr);
+ ob.append(e.isAsc() ? " ASC" : " DESC")
+ .append(OrderByElement.nullsFirst(e.getNullsFirstParam(), e.isAsc()) ?
+ " NULLS FIRST" : " NULLS LAST");
+ orderStrs.add(ob.toString());
+ }
+ sb.append("|OrderExprs:[").append(String.join(",", orderStrs)).append("]");
+
+ sb.append("|Window:")
+ .append(analyticWindow_ == null ? "NONE" : analyticWindow_.toSql());
+ sb.append("|Child:[").append(childKey).append("]");
+ // Analytic predicates are evaluated by a SelectNode above it, so no predicates in
+ // the key.
+ return sb.toString();
+ }
+
@Override
protected String getNodeExplainString(String prefix, String detailPrefix,
TExplainLevel detailLevel) {
diff --git a/fe/src/test/java/org/apache/impala/planner/HboKeyStringTest.java b/fe/src/test/java/org/apache/impala/planner/HboKeyStringTest.java
index c02a39bdfa..63c8c2e5b0 100644
--- a/fe/src/test/java/org/apache/impala/planner/HboKeyStringTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/HboKeyStringTest.java
@@ -1457,4 +1457,221 @@ public class HboKeyStringTest extends FrontendTestBase {
}
assertTopNStorePath(local, merge);
}
+
+ private AnalyticEvalNode singleAnalytic(String query) throws ImpalaException {
+ Map<Integer, PlanNode> nodes = collectPlanNodesInDistributedPlan(query);
+ int count = 0;
+ AnalyticEvalNode analytic = null;
+ for (PlanNode n : nodes.values()) {
+ if (n instanceof AnalyticEvalNode) {
+ analytic = (AnalyticEvalNode) n;
+ count++;
+ }
+ }
+ assertEquals("Expected exactly one AnalyticEvalNode for query: " + query, 1, count);
+ return analytic;
+ }
+
+ @Test
+ public void testAnalyticNodeKeys() throws ImpalaException {
+ String query = "select id, rank() over (partition by int_col order by id) "
+ + "from functional.alltypes where year = 2009";
+ AnalyticEvalNode analytic = singleAnalytic(query);
+
+ String prefix = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|PartExprs:[int_col]|"
+ + "OrderExprs:[id ASC NULLS LAST]|"
+ + "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|";
+ String expectedER = prefix
+ + "Child:[CARDINALITY:ScanNode:functional.alltypes|`year` = 2009]";
+ String expectedIPC = prefix
+ + "Child:[CARDINALITY:ScanNode:functional.alltypes|`year`=<CONST>]";
+
+ assertEquals(expectedER, analytic.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ assertEquals(expectedIPC, analytic.generateHboKeyString(
+ THboStatsType.CARDINALITY,
+ CanonicalizationStrategy.IGNORE_PARTITION_CONSTANTS));
+ }
+
+ @Test
+ public void testAnalyticNodeLimit() throws ImpalaException {
+ // Same analytic as testAnalyticNodeKeys but with a query-level LIMIT, which lands on
+ // the topmost plan node (the AnalyticEvalNode).
+ String query = "select id, rank() over (partition by int_col order by id) "
+ + "from functional.alltypes where year = 2009 limit 5";
+ AnalyticEvalNode analytic = singleAnalytic(query);
+ assertTrue("Expected a valid LIMIT (> 0) on the AnalyticEvalNode",
+ analytic.getLimit() > 0);
+
+ // The LIMIT is also pushed into the analytic's underlying sort so it's a TOPN.
+ String childSortFmt = "CARDINALITY:SortNode:Type:TOPN|IncludeTies:false|"
+ + "OrderKind:LEXICAL|Limit:5|Offset:0|"
+ + "OrderExprs:[int_col ASC NULLS LAST,id ASC NULLS LAST]|"
+ + "Child:[CARDINALITY:ScanNode:functional.alltypes|%s]";
+ String prefix = "CARDINALITY:AnalyticEvalNode:|limit:5|Fns:[rank()]|"
+ + "PartExprs:[int_col]|OrderExprs:[id ASC NULLS LAST]|"
+ + "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|Child:[%s]";
+ String expectedER =
+ String.format(prefix, String.format(childSortFmt, "`year` = 2009"));
+ String expectedIPC =
+ String.format(prefix, String.format(childSortFmt, "`year`=<CONST>"));
+
+ assertEquals(expectedER, analytic.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ assertEquals(expectedIPC, analytic.generateHboKeyString(
+ THboStatsType.CARDINALITY,
+ CanonicalizationStrategy.IGNORE_PARTITION_CONSTANTS));
+ }
+
+ @Test
+ public void testAnalyticImplicitAndExplicitNullsMatch() throws ImpalaException {
+ // A plain "ORDER BY id" (ASC) and an explicit "ORDER BY id ASC NULLS LAST" are
+ // identical: NULLS LAST is the default for ASC.
+ String implicitQuery = "select id, rank() over (partition by int_col order by id) "
+ + "from functional.alltypes where year = 2009";
+ String explicitQuery = "select id, rank() over (partition by int_col order by id "
+ + "asc nulls last) from functional.alltypes where year = 2009";
+ AnalyticEvalNode implicitAnalytic = singleAnalytic(implicitQuery);
+ AnalyticEvalNode explicitAnalytic = singleAnalytic(explicitQuery);
+
+ String prefix = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|PartExprs:[int_col]|"
+ + "OrderExprs:[id ASC NULLS LAST]|"
+ + "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|";
+ String expectedER = prefix
+ + "Child:[CARDINALITY:ScanNode:functional.alltypes|`year` = 2009]";
+ String expectedIPC = prefix
+ + "Child:[CARDINALITY:ScanNode:functional.alltypes|`year`=<CONST>]";
+
+ // Both queries pin the same expected key under each canonicalization strategy.
+ assertEquals(expectedER, implicitAnalytic.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ assertEquals(expectedER, explicitAnalytic.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ assertEquals(expectedIPC, implicitAnalytic.generateHboKeyString(
+ THboStatsType.CARDINALITY,
+ CanonicalizationStrategy.IGNORE_PARTITION_CONSTANTS));
+ assertEquals(expectedIPC, explicitAnalytic.generateHboKeyString(
+ THboStatsType.CARDINALITY,
+ CanonicalizationStrategy.IGNORE_PARTITION_CONSTANTS));
+
+ for (CanonicalizationStrategy strategy : CanonicalizationStrategy.values()) {
+ assertEquals("Implicit ASC and explicit ASC NULLS LAST must produce the same "
+ + "AnalyticEvalNode key for " + strategy,
+ implicitAnalytic.generateHboKeyString(THboStatsType.CARDINALITY, strategy),
+ explicitAnalytic.generateHboKeyString(THboStatsType.CARDINALITY, strategy));
+ }
+ }
+
+ @Test
+ public void testAnalyticPartitionAndOrderAffectKey() throws ImpalaException {
+ // The partition-by and order-by exprs are part of the key, so analytic nodes that
+ // differ only in partitioning or ordering must hash differently.
+ String base = "select id, rank() over (partition by %s order by %s) "
+ + "from functional.alltypes where year = 2009";
+ AnalyticEvalNode byIntCol = singleAnalytic(String.format(base, "int_col", "id"));
+ AnalyticEvalNode byBigintCol =
+ singleAnalytic(String.format(base, "bigint_col", "id"));
+ AnalyticEvalNode orderDesc =
+ singleAnalytic(String.format(base, "int_col", "id desc"));
+
+ for (CanonicalizationStrategy strategy : CanonicalizationStrategy.values()) {
+ String kIntCol =
+ byIntCol.generateHboKeyString(THboStatsType.CARDINALITY, strategy);
+ String kBigintCol =
+ byBigintCol.generateHboKeyString(THboStatsType.CARDINALITY, strategy);
+ String kOrderDesc =
+ orderDesc.generateHboKeyString(THboStatsType.CARDINALITY, strategy);
+ assertNotEquals("Different PARTITION BY must hash differently for " + strategy,
+ kIntCol, kBigintCol);
+ assertNotEquals("Different ORDER BY must hash differently for " + strategy,
+ kIntCol, kOrderDesc);
+ }
+ }
+
+ @Test
+ public void testAnalyticOverJoin() throws ImpalaException {
+ // When an AnalyticEvalNode's input is a join, its partition-by / order-by (and
+ // function) columns reference several input operands and must be qualified with the
+ // join's canonical operand index. Swapping which operand supplies the partition vs.
+ // order column must therefore hash differently.
+ String q1 = "select a.id, rank() over (partition by a.int_col order by b.id) "
+ + "from functional.alltypes a join functional.alltypestiny b on a.id = b.id";
+ String q2 = "select a.id, rank() over (partition by b.int_col order by a.id) "
+ + "from functional.alltypes a join functional.alltypestiny b on a.id = b.id";
+ AnalyticEvalNode an1 = singleAnalytic(q1);
+ AnalyticEvalNode an2 = singleAnalytic(q2);
+
+ String childKey = "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|"
+ + "Child:[CARDINALITY:JoinNode:INNER|Operands:[CARDINALITY:ScanNode:"
+ + "functional.alltypes,CARDINALITY:ScanNode:functional.alltypestiny]|"
+ + "Predicates:[op0.id=op1.id]]";
+ // q1: partition by a.int_col (alltypes=op0), order by b.id (alltypestiny=op1).
+ String expected1 = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|PartExprs:"
+ + "[op0.int_col]|OrderExprs:[op1.id ASC NULLS LAST]|" + childKey;
+ // q2: partition by b.int_col (alltypestiny=op1), order by a.id (alltypes=op0).
+ String expected2 = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|PartExprs:"
+ + "[op1.int_col]|OrderExprs:[op0.id ASC NULLS LAST]|" + childKey;
+
+ for (CanonicalizationStrategy strategy : CanonicalizationStrategy.values()) {
+ assertEquals(expected1,
+ an1.generateHboKeyString(THboStatsType.CARDINALITY, strategy));
+ assertEquals(expected2,
+ an2.generateHboKeyString(THboStatsType.CARDINALITY, strategy));
+ assertNotEquals("Swapped partition/order operands must hash differently for "
+ + strategy,
+ an1.generateHboKeyString(THboStatsType.CARDINALITY, strategy),
+ an2.generateHboKeyString(THboStatsType.CARDINALITY, strategy));
+ }
+ }
+
+ @Test
+ public void testAggOverAnalyticOverJoin() throws ImpalaException {
+ // Aggregation on top of an AnalyticEvalNode on top of a join, grouping on columns
+ // from different join operands and the analytic result.
+ String query = "select int_col, bigint_col, r, count(*) from ("
+ + "select a.int_col int_col, b.bigint_col bigint_col, "
+ + "rank() over (partition by a.int_col order by b.id) r "
+ + "from functional.alltypes a join functional.alltypestiny b on a.id = b.id"
+ + ") t group by int_col, bigint_col, r";
+
+ // The analytic's partition-by (a.int_col) and order-by (b.id) are qualified by the
+ // join operand index, just like the aggregation's grouping columns below.
+ String analyticChild = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|"
+ + "PartExprs:[op0.int_col]|OrderExprs:[op1.id ASC NULLS LAST]|"
+ + "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|"
+ + "Child:[CARDINALITY:JoinNode:INNER|Operands:[CARDINALITY:ScanNode:"
+ + "functional.alltypes,CARDINALITY:ScanNode:functional.alltypestiny]|"
+ + "Predicates:[op0.id=op1.id]]";
+ String expectedPreagg = "CARDINALITY:AggregationNode:FIRST|Preagg:true|"
+ + "GroupingSet:false|AggClasses:[0:Group:op0.int_col,op1.bigint_col,"
+ + "rank() OVER (PARTITION BY op0.int_col ORDER BY op1.id ASC "
+ + "RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]"
+ + "|Child:[" + analyticChild + "]";
+
+ Map<Integer, PlanNode> nodes = collectPlanNodesInDistributedPlan(query);
+ PlanNode preagg = nodes.get(5);
+ assertEquals(expectedPreagg, preagg.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ }
+
+ @Test
+ public void testAnalyticAsJoinOperand() throws ImpalaException {
+ // Test using the analytic result as the join key
+ String query = "select a.id from functional.alltypes a join ("
+ + "select id, rank() over (partition by int_col order by id) r "
+ + "from functional.alltypestiny) b on a.id = b.r";
+ JoinNode join = singleJoin(query);
+
+ String analyticOperand = "CARDINALITY:AnalyticEvalNode:|Fns:[rank()]|"
+ + "PartExprs:[int_col]|OrderExprs:[id ASC NULLS LAST]|"
+ + "Window:RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW|"
+ + "Child:[CARDINALITY:ScanNode:functional.alltypestiny]";
+ // The predicate (a.id = b.r) should be substituted and qualified to get rid of alias.
+ String expectedER = "CARDINALITY:JoinNode:INNER|Operands:["
+ + ALLTYPES_SCAN_KEY + "," + analyticOperand + "]|"
+ + "Predicates:[op0.id=rank() OVER (PARTITION BY op1.int_col ORDER BY op1.id ASC "
+ + "RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]";
+ assertEquals(expectedER, join.generateHboKeyString(
+ THboStatsType.CARDINALITY, CanonicalizationStrategy.EXPR_REWRITE));
+ }
}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-agg.test b/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-agg.test
new file mode 100644
index 0000000000..75535d3ff3
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-agg.test
@@ -0,0 +1,67 @@
+====
+---- QUERY
+EXPLAIN select count(*), rnk % 10 from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_agg_over_analytic') t
+group by rnk % 10
+---- PLAN
+PLAN-ROOT SINK
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| group by: rank() % 10
+| row-size=16B cardinality=10 (from HBO)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=29B cardinality=620
+|
+04:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: id ASC
+|
+01:SORT
+| order by: id ASC
+| row-size=21B cardinality=620
+|
+00:SCAN HDFS [functional.alltypes]
+ partition predicates: `month` = 1
+ HDFS partitions=2/24 files=2 size=40.32KB
+ predicates: string_col != 'test_agg_over_analytic'
+ row-size=21B cardinality=620 (from HBO)
+====
+---- QUERY
+EXPLAIN select count(*), rnk % 10 from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=2 and string_col != 'test_agg_over_analytic') t
+group by rnk % 10
+---- PLAN
+PLAN-ROOT SINK
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| group by: rank() % 10
+| row-size=16B cardinality=10 (from HBO, key ignores partition constants)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=29B cardinality=620
+|
+04:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: id ASC
+|
+01:SORT
+| order by: id ASC
+| row-size=21B cardinality=620
+|
+00:SCAN HDFS [functional.alltypes]
+ partition predicates: `month` = 2
+ HDFS partitions=2/24 files=2 size=36.51KB
+ predicates: string_col != 'test_agg_over_analytic'
+ row-size=21B cardinality=620 (from HBO, key ignores partition constants)
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-join.test b/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-join.test
new file mode 100644
index 0000000000..675a51f59b
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/QueryTest/hbo-analytic-join.test
@@ -0,0 +1,229 @@
+====
+---- QUERY
+# EXPLAIN the analytic query that has been executed.
+EXPLAIN select straight_join t1.int_col, t1.rnk from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_join_over_analytic') t1,
+ (select int_col, rank() over (order by id desc) as rnk
+ from functional.alltypestiny
+ where month=1 and string_col != 'test_join_over_analytic') t2
+where t1.int_col = t2.int_col and t1.rnk = t2.rnk
+---- PLAN
+PLAN-ROOT SINK
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: int_col = int_col, rank() = rank()
+| row-size=66B cardinality=0 (from HBO)
+|
+|--09:EXCHANGE [UNPARTITIONED]
+| |
+| 05:ANALYTIC
+| | functions: rank()
+| | order by: id DESC
+| | window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| | row-size=33B cardinality=2
+| |
+| 08:MERGING-EXCHANGE [UNPARTITIONED]
+| | order by: id DESC
+| |
+| 04:SORT
+| | order by: id DESC
+| | row-size=25B cardinality=2
+| |
+| 03:SCAN HDFS [functional.alltypestiny]
+| partition predicates: `month` = 1
+| HDFS partitions=1/4 files=1 size=115B
+| predicates: string_col != 'test_join_over_analytic'
+| row-size=25B cardinality=2 (from HBO)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=33B cardinality=620
+|
+07:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: id ASC
+|
+01:SORT
+| order by: id ASC
+| row-size=25B cardinality=620
+|
+00:SCAN HDFS [functional.alltypes]
+ partition predicates: `month` = 1
+ HDFS partitions=2/24 files=2 size=40.32KB
+ predicates: string_col != 'test_join_over_analytic'
+ row-size=25B cardinality=620 (from HBO)
+====
+---- QUERY
+# Change the JOIN order
+EXPLAIN select straight_join t1.int_col, t1.rnk from
+ (select int_col, rank() over (order by id desc) as rnk
+ from functional.alltypestiny
+ where month=1 and string_col != 'test_join_over_analytic') t2,
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_join_over_analytic') t1
+where t1.int_col = t2.int_col and t1.rnk = t2.rnk
+---- PLAN
+PLAN-ROOT SINK
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: int_col = int_col, rank() = rank()
+| row-size=66B cardinality=0 (from HBO)
+|
+|--09:EXCHANGE [UNPARTITIONED]
+| |
+| 05:ANALYTIC
+| | functions: rank()
+| | order by: id ASC
+| | window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| | row-size=33B cardinality=620
+| |
+| 08:MERGING-EXCHANGE [UNPARTITIONED]
+| | order by: id ASC
+| |
+| 04:SORT
+| | order by: id ASC
+| | row-size=25B cardinality=620
+| |
+| 03:SCAN HDFS [functional.alltypes]
+| partition predicates: `month` = 1
+| HDFS partitions=2/24 files=2 size=40.32KB
+| predicates: string_col != 'test_join_over_analytic'
+| row-size=25B cardinality=620 (from HBO)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: id DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=33B cardinality=2
+|
+07:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: id DESC
+|
+01:SORT
+| order by: id DESC
+| row-size=25B cardinality=2
+|
+00:SCAN HDFS [functional.alltypestiny]
+ partition predicates: `month` = 1
+ HDFS partitions=1/4 files=1 size=115B
+ predicates: string_col != 'test_join_over_analytic'
+ row-size=25B cardinality=2 (from HBO)
+====
+---- QUERY
+# Changing the analytic functions, the JOIN node have no HBO stats
+EXPLAIN select straight_join t1.int_col, t1.rnk from
+ (select int_col, rank() over (order by bigint_col, id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_join_over_analytic') t1,
+ (select int_col, rank() over (order by bigint_col, id desc) as rnk
+ from functional.alltypestiny
+ where month=1 and string_col != 'test_join_over_analytic') t2
+where t1.int_col = t2.int_col and t1.rnk = t2.rnk
+---- PLAN
+PLAN-ROOT SINK
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: int_col = int_col, rank() = rank()
+| row-size=82B cardinality=1.24K
+|
+|--09:EXCHANGE [UNPARTITIONED]
+| |
+| 05:ANALYTIC
+| | functions: rank()
+| | order by: bigint_col ASC, id DESC
+| | window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| | row-size=41B cardinality=2
+| |
+| 08:MERGING-EXCHANGE [UNPARTITIONED]
+| | order by: bigint_col ASC, id DESC
+| |
+| 04:SORT
+| | order by: bigint_col ASC, id DESC
+| | row-size=33B cardinality=2
+| |
+| 03:SCAN HDFS [functional.alltypestiny]
+| partition predicates: `month` = 1
+| HDFS partitions=1/4 files=1 size=115B
+| predicates: string_col != 'test_join_over_analytic'
+| row-size=33B cardinality=2 (from HBO)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: bigint_col ASC, id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=41B cardinality=620
+|
+07:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: bigint_col ASC, id ASC
+|
+01:SORT
+| order by: bigint_col ASC, id ASC
+| row-size=33B cardinality=620
+|
+00:SCAN HDFS [functional.alltypes]
+ partition predicates: `month` = 1
+ HDFS partitions=2/24 files=2 size=40.32KB
+ predicates: string_col != 'test_join_over_analytic'
+ row-size=33B cardinality=620 (from HBO)
+====
+---- QUERY
+# Change the partition predicate to month=2
+EXPLAIN select straight_join t1.int_col, t1.rnk from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=2 and string_col != 'test_join_over_analytic') t1,
+ (select int_col, rank() over (order by id desc) as rnk
+ from functional.alltypestiny
+ where month=2 and string_col != 'test_join_over_analytic') t2
+where t1.int_col = t2.int_col and t1.rnk = t2.rnk
+---- PLAN
+PLAN-ROOT SINK
+|
+06:HASH JOIN [INNER JOIN, BROADCAST]
+| hash predicates: int_col = int_col, rank() = rank()
+| row-size=66B cardinality=0 (from HBO, key ignores partition constants)
+|
+|--09:EXCHANGE [UNPARTITIONED]
+| |
+| 05:ANALYTIC
+| | functions: rank()
+| | order by: id DESC
+| | window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| | row-size=33B cardinality=2
+| |
+| 08:MERGING-EXCHANGE [UNPARTITIONED]
+| | order by: id DESC
+| |
+| 04:SORT
+| | order by: id DESC
+| | row-size=25B cardinality=2
+| |
+| 03:SCAN HDFS [functional.alltypestiny]
+| partition predicates: `month` = 2
+| HDFS partitions=1/4 files=1 size=115B
+| predicates: string_col != 'test_join_over_analytic'
+| row-size=25B cardinality=2 (from HBO, key ignores partition constants)
+|
+02:ANALYTIC
+| functions: rank()
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=33B cardinality=620
+|
+07:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: id ASC
+|
+01:SORT
+| order by: id ASC
+| row-size=25B cardinality=620
+|
+00:SCAN HDFS [functional.alltypes]
+ partition predicates: `month` = 2
+ HDFS partitions=2/24 files=2 size=36.51KB
+ predicates: string_col != 'test_join_over_analytic'
+ row-size=25B cardinality=620 (from HBO, key ignores partition constants)
+====
diff --git a/tests/query_test/test_hbo.py b/tests/query_test/test_hbo.py
index d132e69984..4aab9599b6 100644
--- a/tests/query_test/test_hbo.py
+++ b/tests/query_test/test_hbo.py
@@ -460,3 +460,31 @@ class TestHBO(HboTestMixin, ImpalaTestSuite):
where string_col != 'test_analytic_topn') t
where rnk <= 5""")
self._run_hbo_explains('QueryTest/hbo-analytic-topn')
+
+ def test_agg_over_analytic_cardinality(self):
+ """Test an Aggregation over analytic nodes using the analytic results as grouping
+ keys. AnalyticEvalNode won't use HBO stats. But its parent, the AggregationNode,
+ should be able to use HBO stats."""
+ self.client.set_configuration(QUERY_OPTIONS)
+ self.execute_query("""
+ select count(*), rnk % 10 from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_agg_over_analytic') t
+ group by rnk % 10""")
+ self._run_hbo_explains('QueryTest/hbo-analytic-agg')
+
+ def test_join_over_analytic_cardinality(self):
+ """Test a JOIN over analytic nodes joining the analytic results. AnalyticEvalNodes
+ won't use HBO stats. But its parent, the JoinNode should be able to use HBO stats."""
+ self.client.set_configuration(QUERY_OPTIONS)
+ self.execute_query("""
+ select straight_join t1.int_col, t1.rnk from
+ (select int_col, rank() over (order by id asc) as rnk
+ from functional.alltypes
+ where month=1 and string_col != 'test_join_over_analytic') t1,
+ (select int_col, rank() over (order by id desc) as rnk
+ from functional.alltypestiny
+ where month=1 and string_col != 'test_join_over_analytic') t2
+ where t1.int_col = t2.int_col and t1.rnk = t2.rnk""")
+ self._run_hbo_explains('QueryTest/hbo-analytic-join')