You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by kr...@apache.org on 2022/09/12 16:07:04 UTC
[hive] branch master updated: HIVE-26498: Implement MV maintenance with Iceberg sources using full rebuild (Krisztian Kasa, reviewed by Laszlo Pinter)
This is an automated email from the ASF dual-hosted git repository.
krisztiankasa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 23c356f0b7c HIVE-26498: Implement MV maintenance with Iceberg sources using full rebuild (Krisztian Kasa, reviewed by Laszlo Pinter)
23c356f0b7c is described below
commit 23c356f0b7c71e7c8fb9663c2badb83d67345cc0
Author: Krisztian Kasa <ka...@gmail.com>
AuthorDate: Mon Sep 12 18:06:51 2022 +0200
HIVE-26498: Implement MV maintenance with Iceberg sources using full rebuild (Krisztian Kasa, reviewed by Laszlo Pinter)
---
.../iceberg/mr/hive/HiveIcebergStorageHandler.java | 18 +++
.../queries/negative/mv_iceberg_orc_timetravel.q | 5 +
.../src/test/queries/negative/mv_mixed_sources.q | 11 ++
.../src/test/queries/positive/mv_iceberg_orc.q | 41 +++++
.../negative/mv_iceberg_orc_timetravel.q.out | 13 ++
.../test/results/negative/mv_mixed_sources.q.out | 25 ++++
.../src/test/results/positive/mv_iceberg_orc.q.out | 166 +++++++++++++++++++++
.../create/CreateMaterializedViewOperation.java | 5 +-
.../update/MaterializedViewUpdateOperation.java | 5 +-
.../org/apache/hadoop/hive/ql/metadata/Hive.java | 16 +-
.../hive/ql/metadata/HiveStorageHandler.java | 18 +++
.../hive/ql/metadata/MaterializedViewMetadata.java | 30 +++-
.../HiveRelOptMaterializationValidator.java | 3 +-
.../rules/views/HiveMaterializedViewUtils.java | 125 +++++++++++++++-
.../hadoop/hive/ql/parse/SemanticAnalyzer.java | 17 ++-
.../hadoop/hive/metastore/txn/TxnHandler.java | 9 +-
.../hive/common/MaterializationSnapshot.java | 110 ++++++++++++++
.../hadoop/hive/common/type/SnapshotContext.java | 66 ++++++++
18 files changed, 651 insertions(+), 32 deletions(-)
diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
index 51e3d60ad9d..e7dcbf67ef1 100644
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
@@ -38,6 +38,7 @@ import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.common.type.Date;
+import org.apache.hadoop.hive.common.type.SnapshotContext;
import org.apache.hadoop.hive.common.type.Timestamp;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
@@ -89,6 +90,7 @@ import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SortDirection;
import org.apache.iceberg.SortField;
@@ -977,4 +979,20 @@ public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, H
return conf;
}
}
+
+ @Override
+ public boolean areSnapshotsSupported() {
+ return true;
+ }
+
+ @Override
+ public SnapshotContext getCurrentSnapshotContext(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
+ Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
+ Snapshot current = table.currentSnapshot();
+ if (current == null) {
+ return null;
+ }
+ return new SnapshotContext(current.snapshotId());
+ }
}
diff --git a/iceberg/iceberg-handler/src/test/queries/negative/mv_iceberg_orc_timetravel.q b/iceberg/iceberg-handler/src/test/queries/negative/mv_iceberg_orc_timetravel.q
new file mode 100644
index 00000000000..301cc617e68
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/queries/negative/mv_iceberg_orc_timetravel.q
@@ -0,0 +1,5 @@
+drop table if exists tbl_ice;
+create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2');
+
+create materialized view mat1 as
+select b, c from tbl_ice for system_version as of 5422037307753150798;
diff --git a/iceberg/iceberg-handler/src/test/queries/negative/mv_mixed_sources.q b/iceberg/iceberg-handler/src/test/queries/negative/mv_mixed_sources.q
new file mode 100644
index 00000000000..217bfae9167
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/queries/negative/mv_mixed_sources.q
@@ -0,0 +1,11 @@
+set hive.support.concurrency=true;
+set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
+
+drop table if exists tbl_acid;
+create table tbl_acid(a int, b string, c int) stored as orc TBLPROPERTIES ('transactional'='true');
+
+drop table if exists tbl_ice;
+create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2');
+
+create materialized view mat1 as
+select tbl_acid.b, tbl_acid.c from tbl_ice join tbl_acid on tbl_ice.a = tbl_acid.a;
diff --git a/iceberg/iceberg-handler/src/test/queries/positive/mv_iceberg_orc.q b/iceberg/iceberg-handler/src/test/queries/positive/mv_iceberg_orc.q
new file mode 100644
index 00000000000..e040fc8e6fd
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/queries/positive/mv_iceberg_orc.q
@@ -0,0 +1,41 @@
+-- SORT_QUERY_RESULTS
+
+set hive.support.concurrency=true;
+set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
+
+drop table if exists tbl_ice;
+drop table if exists tbl_ice_v2;
+
+create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='1');
+create external table tbl_ice_v2(d int, e string, f int) stored by iceberg stored as orc tblproperties ('format-version'='2');
+
+insert into tbl_ice_v2 values (1, 'one v2', 50), (4, 'four v2', 53), (5, 'five v2', 54);
+
+create materialized view mat1 as
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52;
+
+-- view should be empty
+select * from mat1;
+
+-- view is up-to-date, use it
+explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52;
+
+-- insert some new values to one of the source tables
+insert into tbl_ice values (1, 'one', 50), (2, 'two', 51), (3, 'three', 52), (4, 'four', 53), (5, 'five', 54);
+
+-- view is outdated, can not be used
+explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52;
+
+explain cbo
+alter materialized view mat1 rebuild;
+
+alter materialized view mat1 rebuild;
+
+-- view should contain data
+select * from mat1;
+
+-- view is up-to-date again, use it
+explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52;
diff --git a/iceberg/iceberg-handler/src/test/results/negative/mv_iceberg_orc_timetravel.q.out b/iceberg/iceberg-handler/src/test/results/negative/mv_iceberg_orc_timetravel.q.out
new file mode 100644
index 00000000000..5eeea644791
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/results/negative/mv_iceberg_orc_timetravel.q.out
@@ -0,0 +1,13 @@
+PREHOOK: query: drop table if exists tbl_ice
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: drop table if exists tbl_ice
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@tbl_ice
+POSTHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@tbl_ice
+FAILED: SemanticException Automatic rewriting for materialized view cannot be enabled if the materialized view uses time travel query.
diff --git a/iceberg/iceberg-handler/src/test/results/negative/mv_mixed_sources.q.out b/iceberg/iceberg-handler/src/test/results/negative/mv_mixed_sources.q.out
new file mode 100644
index 00000000000..a8f41cc0db2
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/results/negative/mv_mixed_sources.q.out
@@ -0,0 +1,25 @@
+PREHOOK: query: drop table if exists tbl_acid
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: drop table if exists tbl_acid
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: create table tbl_acid(a int, b string, c int) stored as orc TBLPROPERTIES ('transactional'='true')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@tbl_acid
+POSTHOOK: query: create table tbl_acid(a int, b string, c int) stored as orc TBLPROPERTIES ('transactional'='true')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@tbl_acid
+PREHOOK: query: drop table if exists tbl_ice
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: drop table if exists tbl_ice
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@tbl_ice
+POSTHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@tbl_ice
+FAILED: SemanticException All materialized view source tables either must be native ACID tables or must support table snapshots.
diff --git a/iceberg/iceberg-handler/src/test/results/positive/mv_iceberg_orc.q.out b/iceberg/iceberg-handler/src/test/results/positive/mv_iceberg_orc.q.out
new file mode 100644
index 00000000000..baef62951ac
--- /dev/null
+++ b/iceberg/iceberg-handler/src/test/results/positive/mv_iceberg_orc.q.out
@@ -0,0 +1,166 @@
+PREHOOK: query: drop table if exists tbl_ice
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: drop table if exists tbl_ice
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: drop table if exists tbl_ice_v2
+PREHOOK: type: DROPTABLE
+POSTHOOK: query: drop table if exists tbl_ice_v2
+POSTHOOK: type: DROPTABLE
+PREHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='1')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@tbl_ice
+POSTHOOK: query: create external table tbl_ice(a int, b string, c int) stored by iceberg stored as orc tblproperties ('format-version'='1')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@tbl_ice
+PREHOOK: query: create external table tbl_ice_v2(d int, e string, f int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@tbl_ice_v2
+POSTHOOK: query: create external table tbl_ice_v2(d int, e string, f int) stored by iceberg stored as orc tblproperties ('format-version'='2')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@tbl_ice_v2
+PREHOOK: query: insert into tbl_ice_v2 values (1, 'one v2', 50), (4, 'four v2', 53), (5, 'five v2', 54)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@tbl_ice_v2
+POSTHOOK: query: insert into tbl_ice_v2 values (1, 'one v2', 50), (4, 'four v2', 53), (5, 'five v2', 54)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@tbl_ice_v2
+PREHOOK: query: create materialized view mat1 as
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+PREHOOK: type: CREATE_MATERIALIZED_VIEW
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: database:default
+PREHOOK: Output: default@mat1
+POSTHOOK: query: create materialized view mat1 as
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+POSTHOOK: type: CREATE_MATERIALIZED_VIEW
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@mat1
+POSTHOOK: Lineage: mat1.b SIMPLE [(tbl_ice)tbl_ice.FieldSchema(name:b, type:string, comment:null), ]
+POSTHOOK: Lineage: mat1.c SIMPLE [(tbl_ice)tbl_ice.FieldSchema(name:c, type:int, comment:null), ]
+POSTHOOK: Lineage: mat1.e SIMPLE [(tbl_ice_v2)tbl_ice_v2.FieldSchema(name:e, type:string, comment:null), ]
+PREHOOK: query: select * from mat1
+PREHOOK: type: QUERY
+PREHOOK: Input: default@mat1
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select * from mat1
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@mat1
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+PREHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+PREHOOK: type: QUERY
+PREHOOK: Input: default@mat1
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@mat1
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+CBO PLAN:
+HiveTableScan(table=[[default, mat1]], table:alias=[default.mat1])
+
+PREHOOK: query: insert into tbl_ice values (1, 'one', 50), (2, 'two', 51), (3, 'three', 52), (4, 'four', 53), (5, 'five', 54)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@tbl_ice
+POSTHOOK: query: insert into tbl_ice values (1, 'one', 50), (2, 'two', 51), (3, 'three', 52), (4, 'four', 53), (5, 'five', 54)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@tbl_ice
+PREHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+PREHOOK: type: QUERY
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+CBO PLAN:
+HiveProject(b=[$1], c=[$2], e=[$4])
+ HiveJoin(condition=[=($0, $3)], joinType=[inner], algorithm=[none], cost=[not available])
+ HiveProject(a=[$0], b=[$1], c=[$2])
+ HiveFilter(condition=[AND(>($2, 52), IS NOT NULL($0))])
+ HiveTableScan(table=[[default, tbl_ice]], table:alias=[tbl_ice])
+ HiveProject(d=[$0], e=[$1])
+ HiveFilter(condition=[IS NOT NULL($0)])
+ HiveTableScan(table=[[default, tbl_ice_v2]], table:alias=[tbl_ice_v2])
+
+PREHOOK: query: explain cbo
+alter materialized view mat1 rebuild
+PREHOOK: type: ALTER_MATERIALIZED_VIEW_REBUILD
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: default@mat1
+POSTHOOK: query: explain cbo
+alter materialized view mat1 rebuild
+POSTHOOK: type: ALTER_MATERIALIZED_VIEW_REBUILD
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: default@mat1
+CBO PLAN:
+HiveProject(b=[$1], c=[$2], e=[$4])
+ HiveJoin(condition=[=($0, $3)], joinType=[inner], algorithm=[none], cost=[not available])
+ HiveProject(a=[$0], b=[$1], c=[$2])
+ HiveFilter(condition=[AND(>($2, 52), IS NOT NULL($0))])
+ HiveTableScan(table=[[default, tbl_ice]], table:alias=[tbl_ice])
+ HiveProject(d=[$0], e=[$1])
+ HiveFilter(condition=[IS NOT NULL($0)])
+ HiveTableScan(table=[[default, tbl_ice_v2]], table:alias=[tbl_ice_v2])
+
+PREHOOK: query: alter materialized view mat1 rebuild
+PREHOOK: type: ALTER_MATERIALIZED_VIEW_REBUILD
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: default@mat1
+POSTHOOK: query: alter materialized view mat1 rebuild
+POSTHOOK: type: ALTER_MATERIALIZED_VIEW_REBUILD
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: default@mat1
+POSTHOOK: Lineage: mat1.b SIMPLE [(tbl_ice)tbl_ice.FieldSchema(name:b, type:string, comment:null), ]
+POSTHOOK: Lineage: mat1.c SIMPLE [(tbl_ice)tbl_ice.FieldSchema(name:c, type:int, comment:null), ]
+POSTHOOK: Lineage: mat1.e SIMPLE [(tbl_ice_v2)tbl_ice_v2.FieldSchema(name:e, type:string, comment:null), ]
+PREHOOK: query: select * from mat1
+PREHOOK: type: QUERY
+PREHOOK: Input: default@mat1
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: select * from mat1
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@mat1
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+five 54 five v2
+four 53 four v2
+PREHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+PREHOOK: type: QUERY
+PREHOOK: Input: default@mat1
+PREHOOK: Input: default@tbl_ice
+PREHOOK: Input: default@tbl_ice_v2
+PREHOOK: Output: hdfs://### HDFS PATH ###
+POSTHOOK: query: explain cbo
+select tbl_ice.b, tbl_ice.c, tbl_ice_v2.e from tbl_ice join tbl_ice_v2 on tbl_ice.a=tbl_ice_v2.d where tbl_ice.c > 52
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@mat1
+POSTHOOK: Input: default@tbl_ice
+POSTHOOK: Input: default@tbl_ice_v2
+POSTHOOK: Output: hdfs://### HDFS PATH ###
+CBO PLAN:
+HiveTableScan(table=[[default, mat1]], table:alias=[default.mat1])
+
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/create/CreateMaterializedViewOperation.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/create/CreateMaterializedViewOperation.java
index b9f2216cac3..d8aa28158c0 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/create/CreateMaterializedViewOperation.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/create/CreateMaterializedViewOperation.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hive.ql.ddl.view.create;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.TableName;
-import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.SourceTable;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
@@ -41,6 +40,8 @@ import java.util.Set;
import org.apache.hadoop.hive.ql.io.AcidUtils;
+import static org.apache.hadoop.hive.ql.optimizer.calcite.rules.views.HiveMaterializedViewUtils.getSnapshotOf;
+
/**
* Operation process of creating a view.
*/
@@ -74,7 +75,7 @@ public class CreateMaterializedViewOperation extends DDLOperation<CreateMaterial
tbl.getDbName(),
tbl.getTableName(),
sourceTables,
- context.getConf().get(ValidTxnWriteIdList.VALID_TABLES_WRITEIDS_KEY));
+ getSnapshotOf(context, desc.getTablesUsed()));
tbl.setMaterializedViewMetadata(metadata);
}
context.getDb().createTable(tbl, desc.getIfNotExists());
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/materialized/update/MaterializedViewUpdateOperation.java b/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/materialized/update/MaterializedViewUpdateOperation.java
index 9e91da89059..ef91a7aff7d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/materialized/update/MaterializedViewUpdateOperation.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/ddl/view/materialized/update/MaterializedViewUpdateOperation.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.hive.ql.ddl.view.materialized.update;
-import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
import org.apache.hadoop.hive.ql.ddl.DDLOperation;
import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
import org.apache.hadoop.hive.ql.metadata.HiveException;
@@ -27,6 +26,8 @@ import org.apache.hadoop.hive.ql.metadata.MaterializedViewMetadata;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.parse.ExplainConfiguration.AnalyzeState;
+import static org.apache.hadoop.hive.ql.optimizer.calcite.rules.views.HiveMaterializedViewUtils.getSnapshotOf;
+
/**
* Operation process of updating a materialized view.
*/
@@ -53,7 +54,7 @@ public class MaterializedViewUpdateOperation extends DDLOperation<MaterializedVi
// We need to update the status of the creation signature
Table mvTable = context.getDb().getTable(desc.getName());
MaterializedViewMetadata newMetadata = mvTable.getMVMetadata().reset(
- context.getConf().get(ValidTxnWriteIdList.VALID_TABLES_WRITEIDS_KEY));
+ getSnapshotOf(context, mvTable.getMVMetadata().getSourceTableNames()));
context.getDb().updateCreationMetadata(mvTable.getDbName(), mvTable.getTableName(), newMetadata);
mvTable.setMaterializedViewMetadata(newMetadata);
HiveMaterializedViewsRegistry.get().refreshMaterializedView(context.getDb().getConf(), mvTable);
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
index 50e08fdb93c..a6f38589fce 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
@@ -98,7 +98,6 @@ import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.common.TableName;
import org.apache.hadoop.hive.common.ValidReaderWriteIdList;
import org.apache.hadoop.hive.common.ValidTxnList;
-import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.hadoop.hive.common.classification.InterfaceAudience.LimitedPrivate;
import org.apache.hadoop.hive.common.classification.InterfaceStability.Unstable;
@@ -1968,8 +1967,7 @@ public class Hive {
// if rewriting with outdated materialized views is enabled (currently
// disabled by default).
materialization = HiveMaterializedViewUtils.augmentMaterializationWithTimeInformation(
- materialization, validTxnsList, new ValidTxnWriteIdList(
- materializedViewTable.getMVMetadata().getValidTxnList()));
+ materialization, validTxnsList, materializedViewTable.getMVMetadata().getSnapshot());
}
result.addAll(HiveMaterializedViewUtils.deriveGroupingSetsMaterializedViews(materialization));
}
@@ -1987,7 +1985,7 @@ public class Hive {
*/
public Boolean isOutdatedMaterializedView(
Table materializedViewTable, Set<TableName> tablesUsed,
- boolean forceMVContentsUpToDate, HiveTxnManager txnMgr) throws LockException {
+ boolean forceMVContentsUpToDate, HiveTxnManager txnMgr) throws HiveException {
String validTxnsList = conf.get(ValidTxnList.VALID_TXNS_KEY);
if (validTxnsList == null) {
@@ -2013,7 +2011,7 @@ public class Hive {
if (forceMVContentsUpToDate || timeWindow == 0L ||
mvMetadata.getMaterializationTime() < System.currentTimeMillis() - timeWindow) {
return HiveMaterializedViewUtils.isOutdatedMaterializedView(
- validTxnsList, txnMgr, tablesUsed, materializedViewTable);
+ validTxnsList, txnMgr, this, tablesUsed, materializedViewTable);
}
}
return outdated;
@@ -2034,7 +2032,7 @@ public class Hive {
}
return HiveMaterializedViewUtils.isOutdatedMaterializedView(
- validTxnsList, txnManager, table.getMVMetadata().getSourceTableNames(), table);
+ validTxnsList, txnManager, this, table.getMVMetadata().getSourceTableNames(), table);
}
/**
@@ -2201,8 +2199,7 @@ public class Hive {
// We will rewrite it to include the filters on transaction list
// so we can produce partial rewritings
relOptMaterialization = HiveMaterializedViewUtils.augmentMaterializationWithTimeInformation(
- relOptMaterialization, validTxnsList, new ValidTxnWriteIdList(
- metadata.getValidTxnList()));
+ relOptMaterialization, validTxnsList, metadata.getSnapshot());
}
addToMaterializationList(expandGroupingSets, invalidationInfo, relOptMaterialization, result);
continue;
@@ -2224,8 +2221,7 @@ public class Hive {
// We will rewrite it to include the filters on transaction list
// so we can produce partial rewritings
relOptMaterialization = HiveMaterializedViewUtils.augmentMaterializationWithTimeInformation(
- hiveRelOptMaterialization, validTxnsList, new ValidTxnWriteIdList(
- metadata.getValidTxnList()));
+ hiveRelOptMaterialization, validTxnsList, metadata.getSnapshot());
}
addToMaterializationList(expandGroupingSets, invalidationInfo, relOptMaterialization, result);
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
index b67d5d83476..65e458cc650 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
@@ -26,6 +26,7 @@ import java.util.Collections;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.hive.common.classification.InterfaceAudience;
import org.apache.hadoop.hive.common.classification.InterfaceStability;
+import org.apache.hadoop.hive.common.type.SnapshotContext;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
@@ -489,4 +490,21 @@ public interface HiveStorageHandler extends Configurable {
*/
default void executeOperation(org.apache.hadoop.hive.ql.metadata.Table table, AlterTableExecuteSpec executeSpec) {
}
+
+ /**
+ * Gets whether this storage handler supports snapshots.
+ * @return true means snapshots are supported false otherwise
+ */
+ default boolean areSnapshotsSupported() {
+ return false;
+ }
+
+ /**
+ * Query the most recent unique snapshot's context of the passed table.
+ * @param table - {@link org.apache.hadoop.hive.ql.metadata.Table} which snapshot context should be returned.
+ * @return {@link SnapshotContext} wraps the snapshotId or null if no snapshot present.
+ */
+ default SnapshotContext getCurrentSnapshotContext(org.apache.hadoop.hive.ql.metadata.Table table) {
+ return null;
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/MaterializedViewMetadata.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/MaterializedViewMetadata.java
index 12bfd38e80c..47a92df630d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/MaterializedViewMetadata.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/MaterializedViewMetadata.java
@@ -18,9 +18,12 @@
package org.apache.hadoop.hive.ql.metadata;
+import org.apache.hadoop.hive.common.MaterializationSnapshot;
import org.apache.hadoop.hive.common.TableName;
import org.apache.hadoop.hive.metastore.api.CreationMetadata;
import org.apache.hadoop.hive.metastore.api.SourceTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
@@ -32,6 +35,8 @@ import static java.util.Collections.unmodifiableList;
import static java.util.Collections.unmodifiableSet;
public class MaterializedViewMetadata {
+ private static final Logger LOG = LoggerFactory.getLogger(MaterializedViewMetadata.class);
+
final CreationMetadata creationMetadata;
MaterializedViewMetadata(CreationMetadata creationMetadata) {
@@ -39,9 +44,10 @@ public class MaterializedViewMetadata {
}
public MaterializedViewMetadata(
- String catalogName, String dbName, String mvName, Set<SourceTable> sourceTables, String validTxnList) {
+ String catalogName, String dbName, String mvName, Set<SourceTable> sourceTables,
+ MaterializationSnapshot snapshot) {
this.creationMetadata = new CreationMetadata(catalogName, dbName, mvName, toFullTableNames(sourceTables));
- this.creationMetadata.setValidTxnList(validTxnList);
+ this.creationMetadata.setValidTxnList(snapshot.asJsonString());
this.creationMetadata.setSourceTables(unmodifiableList(new ArrayList<>(sourceTables)));
}
@@ -81,15 +87,27 @@ public class MaterializedViewMetadata {
return unmodifiableList(creationMetadata.getSourceTables());
}
- public String getValidTxnList() {
- return creationMetadata.getValidTxnList();
+ /**
+ * Get the snapshot data of the materialized view source tables stored in HMS.
+ * The returned object contains the valid txn list in case of native acid tables
+ * or the snapshotIds of tables which storage handle supports snapshots.
+ * @return {@link MaterializationSnapshot} object or null if the wrapped {@link CreationMetadata} object
+ * does not contain snapshot data
+ */
+ public MaterializationSnapshot getSnapshot() {
+ if (creationMetadata.getValidTxnList() == null || creationMetadata.getValidTxnList().isEmpty()) {
+ LOG.debug("Could not obtain materialization snapshot of materialized view {}.{}",
+ creationMetadata.getDbName(), creationMetadata.getTblName());
+ return null;
+ }
+ return MaterializationSnapshot.fromJson(creationMetadata.getValidTxnList());
}
public long getMaterializationTime() {
return creationMetadata.getMaterializationTime();
}
- public MaterializedViewMetadata reset(String validTxnList) {
+ public MaterializedViewMetadata reset(MaterializationSnapshot snapshot) {
Set<SourceTable> newSourceTables =
creationMetadata.getSourceTables().stream().map(this::from).collect(Collectors.toSet());
@@ -98,7 +116,7 @@ public class MaterializedViewMetadata {
creationMetadata.getDbName(),
creationMetadata.getTblName(),
unmodifiableSet(newSourceTables),
- validTxnList);
+ snapshot);
}
private SourceTable from(SourceTable sourceTable) {
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveRelOptMaterializationValidator.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveRelOptMaterializationValidator.java
index aeedc6bd40b..3897a2912a6 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveRelOptMaterializationValidator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveRelOptMaterializationValidator.java
@@ -38,6 +38,7 @@ import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.Util;
import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveAggregate;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveAntiJoin;
@@ -88,7 +89,7 @@ public class HiveRelOptMaterializationValidator extends HiveRelShuttleImpl {
if (tab.isTemporary()) {
fail(tab.getTableName() + " is a temporary table");
}
- if (tab.getTableType() == TableType.EXTERNAL_TABLE) {
+ if (tab.getTableType() == TableType.EXTERNAL_TABLE && !AcidUtils.isNonNativeAcidTable(tab)) {
fail(tab.getFullyQualifiedName() + " is an external table");
}
return hiveScan;
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/views/HiveMaterializedViewUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/views/HiveMaterializedViewUtils.java
index cde22b3f0c5..4767b08f841 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/views/HiveMaterializedViewUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/views/HiveMaterializedViewUtils.java
@@ -21,7 +21,10 @@ import com.google.common.collect.ImmutableList;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
@@ -49,11 +52,16 @@ import org.apache.calcite.util.ImmutableBitSet;
import org.apache.hadoop.hive.common.TableName;
import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
+import org.apache.hadoop.hive.common.type.SnapshotContext;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
import org.apache.hadoop.hive.ql.lockmgr.LockException;
+import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveRelOptMaterialization;
+import org.apache.hadoop.hive.common.MaterializationSnapshot;
+import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.MaterializedViewMetadata;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.calcite.HiveRelFactories;
@@ -98,7 +106,23 @@ public class HiveMaterializedViewUtils {
* materialized view definition uses external tables.
*/
public static Boolean isOutdatedMaterializedView(
- String validTxnsList, HiveTxnManager txnMgr,
+ String validTxnsList, HiveTxnManager txnMgr, Hive db,
+ Set<TableName> tablesUsed, Table materializedViewTable) throws HiveException {
+
+ MaterializedViewMetadata mvMetadata = materializedViewTable.getMVMetadata();
+ MaterializationSnapshot snapshot = mvMetadata.getSnapshot();
+
+ if (snapshot != null && snapshot.getTableSnapshots() != null && !snapshot.getTableSnapshots().isEmpty()) {
+ return isOutdatedMaterializedView(snapshot, db, tablesUsed, materializedViewTable);
+ }
+
+ String materializationTxnList = snapshot != null ? snapshot.getValidTxnList() : null;
+ return isOutdatedMaterializedView(
+ materializationTxnList, validTxnsList, txnMgr, tablesUsed, materializedViewTable);
+ }
+
+ private static Boolean isOutdatedMaterializedView(
+ String materializationTxnList, String validTxnsList, HiveTxnManager txnMgr,
Set<TableName> tablesUsed, Table materializedViewTable) throws LockException {
List<String> tablesUsedNames = tablesUsed.stream()
.map(tableName -> TableName.getDbTable(tableName.getDb(), tableName.getTable()))
@@ -110,17 +134,14 @@ public class HiveMaterializedViewUtils {
return null;
}
- MaterializedViewMetadata mvMetadata = materializedViewTable.getMVMetadata();
Set<String> storedTablesUsed = materializedViewTable.getMVMetadata().getSourceTableFullNames();
- if (mvMetadata.getValidTxnList() == null ||
- mvMetadata.getValidTxnList().isEmpty()) {
+ if (materializationTxnList == null || materializationTxnList.isEmpty()) {
LOG.debug("Materialized view " + materializedViewTable.getFullyQualifiedName() +
" ignored for rewriting as we could not obtain materialization txn ids");
return null;
}
+ ValidTxnWriteIdList materializationTxnWriteIds = new ValidTxnWriteIdList(materializationTxnList);
boolean ignore = false;
- ValidTxnWriteIdList mvTxnWriteIds = new ValidTxnWriteIdList(
- mvMetadata.getValidTxnList());
for (String fullyQualifiedTableName : tablesUsedNames) {
// Note. If the materialized view does not contain a table that is contained in the query,
// we do not need to check whether that specific table is outdated or not. If a rewriting
@@ -138,12 +159,12 @@ public class HiveMaterializedViewUtils {
ignore = true;
break;
}
- ValidWriteIdList tableWriteIds = mvTxnWriteIds.getTableValidWriteIdList(fullyQualifiedTableName);
+ ValidWriteIdList tableWriteIds = materializationTxnWriteIds.getTableValidWriteIdList(fullyQualifiedTableName);
if (tableWriteIds == null) {
// This should not happen, but we ignore for safety
LOG.warn("Materialized view " + materializedViewTable.getFullyQualifiedName() +
" ignored for rewriting as details about txn ids for table " + fullyQualifiedTableName +
- " could not be found in " + mvTxnWriteIds);
+ " could not be found in " + materializationTxnWriteIds);
ignore = true;
break;
}
@@ -160,11 +181,72 @@ public class HiveMaterializedViewUtils {
return false;
}
+ private static Boolean isOutdatedMaterializedView(
+ MaterializationSnapshot snapshot, Hive db,
+ Set<TableName> tablesUsed, Table materializedViewTable) throws HiveException {
+ List<String> tablesUsedNames = tablesUsed.stream()
+ .map(tableName -> TableName.getDbTable(tableName.getDb(), tableName.getTable()))
+ .collect(Collectors.toList());
+
+ Map<String, SnapshotContext> snapshotMap = snapshot.getTableSnapshots();
+ if (snapshotMap == null || snapshotMap.isEmpty()) {
+ LOG.debug("Materialized view {} ignored for rewriting as we could not obtain current snapshot ids",
+ materializedViewTable.getFullyQualifiedName());
+ return null;
+ }
+
+ Set<String> storedTablesUsed = materializedViewTable.getMVMetadata().getSourceTableFullNames();
+ for (String fullyQualifiedTableName : tablesUsedNames) {
+ // Note. If the materialized view does not contain a table that is contained in the query,
+ // we do not need to check whether that specific table is outdated or not. If a rewriting
+ // is produced in those cases, it is because that additional table is joined with the
+ // existing tables with an append-columns only join, i.e., PK-FK + not null.
+ if (!storedTablesUsed.contains(fullyQualifiedTableName)) {
+ continue;
+ }
+
+ Table table = db.getTable(fullyQualifiedTableName);
+ if (table.getStorageHandler() == null) {
+ LOG.debug("Materialized view {} ignored for rewriting as we could not get storage handler of table {}",
+ materializedViewTable.getFullyQualifiedName(), fullyQualifiedTableName);
+ return null;
+ }
+ if (!table.getStorageHandler().areSnapshotsSupported()) {
+ LOG.debug("Materialized view {} ignored for rewriting as storage handler of table {} " +
+ "does not support snapshots.",
+ materializedViewTable.getFullyQualifiedName(), fullyQualifiedTableName);
+ return null;
+ }
+ SnapshotContext currentTableSnapshot = table.getStorageHandler().getCurrentSnapshotContext(table);
+ SnapshotContext storedTableSnapshot = snapshotMap.get(fullyQualifiedTableName);
+ if (!Objects.equals(currentTableSnapshot, storedTableSnapshot)) {
+ LOG.debug("Materialized view {} contents are outdated", materializedViewTable.getFullyQualifiedName());
+ return true;
+ }
+ }
+
+ return false;
+ }
+
/**
* Method to enrich the materialization query contained in the input with
* its invalidation.
*/
public static HiveRelOptMaterialization augmentMaterializationWithTimeInformation(
+ HiveRelOptMaterialization materialization, String validTxnsList,
+ MaterializationSnapshot snapshot) throws LockException {
+
+ if (snapshot != null && snapshot.getTableSnapshots() != null && !snapshot.getTableSnapshots().isEmpty()) {
+ // Not supported yet for Iceberg tables
+ return materialization;
+ }
+
+ String materializationTxnList = snapshot != null ? snapshot.getValidTxnList() : null;
+ return augmentMaterializationWithTimeInformation(
+ materialization, validTxnsList, new ValidTxnWriteIdList(materializationTxnList));
+ }
+
+ private static HiveRelOptMaterialization augmentMaterializationWithTimeInformation(
HiveRelOptMaterialization materialization, String validTxnsList,
ValidTxnWriteIdList materializationTxnList) throws LockException {
// Extract tables used by the query which will in turn be used to generate
@@ -403,4 +485,31 @@ public class HiveMaterializedViewUtils {
}
return newScan;
}
+
+ public static MaterializationSnapshot getSnapshotOf(DDLOperationContext context, Set<TableName> tables)
+ throws HiveException {
+ Map<String, SnapshotContext> snapshot = getSnapshotOf(context.getDb(), tables);
+ if (snapshot.isEmpty()) {
+ return new MaterializationSnapshot(context.getConf().get(ValidTxnWriteIdList.VALID_TABLES_WRITEIDS_KEY));
+ }
+
+ return new MaterializationSnapshot(snapshot);
+ }
+
+ private static Map<String, SnapshotContext> getSnapshotOf(Hive db, Set<TableName> tables) throws HiveException {
+ Map<String, SnapshotContext> snapshot = new HashMap<>(tables.size());
+ for (TableName tableName : tables) {
+ Table table = db.getTable(tableName);
+ if (table.getStorageHandler() != null) {
+ HiveStorageHandler storageHandler = table.getStorageHandler();
+ if (!storageHandler.areSnapshotsSupported()) {
+ return Collections.emptyMap();
+ }
+ snapshot.put(table.getFullyQualifiedName(), storageHandler.getCurrentSnapshotContext(table));
+ } else {
+ return Collections.emptyMap();
+ }
+ }
+ return snapshot;
+ }
}
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index 7afea4a222b..ce498a869f3 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql.parse;
import static java.util.Objects.nonNull;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.hadoop.hive.common.AcidConstants.SOFT_DELETE_TABLE;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.DYNAMICPARTITIONCONVERT;
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVEARCHIVEENABLED;
@@ -14212,15 +14213,29 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
// Do not allow view to be defined on temp table or other materialized view
validateTablesUsed(this);
if (createVwDesc.isRewriteEnabled()) {
+ int nativeAcidCount = 0;
+ int supportsSnapshotCount = 0;
for (TableScanOperator ts : topOps.values()) {
Table table = ts.getConf().getTableMetadata();
if (SemanticAnalyzer.DUMMY_TABLE.equals(table.getTableName())) {
continue;
}
- if (!AcidUtils.isTransactionalTable(table)) {
+ if (AcidUtils.isTransactionalTable(table)) {
+ ++nativeAcidCount;
+ } else if (AcidUtils.isNonNativeAcidTable(table) && table.getStorageHandler().areSnapshotsSupported()) {
+ ++supportsSnapshotCount;
+ } else {
throw new SemanticException("Automatic rewriting for materialized view cannot "
+ "be enabled if the materialized view uses non-transactional tables");
}
+ if (isNotBlank(ts.getConf().getAsOfTimestamp()) || isNotBlank(ts.getConf().getAsOfVersion())) {
+ throw new SemanticException("Automatic rewriting for materialized view cannot "
+ + "be enabled if the materialized view uses time travel query.");
+ }
+ }
+ if (nativeAcidCount > 0 && supportsSnapshotCount > 0) {
+ throw new SemanticException("All materialized view source tables either must be native ACID tables or " +
+ "must support table snapshots.");
}
}
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
index 4d49df97315..226aa269147 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
@@ -68,6 +68,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.MaterializationSnapshot;
import org.apache.hadoop.hive.common.TableName;
import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
import org.apache.hadoop.hive.common.ValidReadTxnList;
@@ -2572,8 +2573,12 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
// the materialization was created. Otherwise, query returns 0 rows.
// Parse validReaderWriteIdList from creation metadata
- final ValidTxnWriteIdList validReaderWriteIdList =
- new ValidTxnWriteIdList(creationMetadata.getValidTxnList());
+ MaterializationSnapshot mvSnapshot = MaterializationSnapshot.fromJson(creationMetadata.getValidTxnList());
+ if (mvSnapshot.getTableSnapshots() != null && !mvSnapshot.getTableSnapshots().isEmpty()) {
+ // Incremental rebuild of MVs on Iceberg sources is not supported.
+ return null;
+ }
+ final ValidTxnWriteIdList validReaderWriteIdList = new ValidTxnWriteIdList(mvSnapshot.getValidTxnList());
// Parse validTxnList
final ValidReadTxnList currentValidTxnList = new ValidReadTxnList(validTxnListStr);
diff --git a/storage-api/src/java/org/apache/hadoop/hive/common/MaterializationSnapshot.java b/storage-api/src/java/org/apache/hadoop/hive/common/MaterializationSnapshot.java
new file mode 100644
index 00000000000..b1bbba5c2bf
--- /dev/null
+++ b/storage-api/src/java/org/apache/hadoop/hive/common/MaterializationSnapshot.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.common;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hadoop.hive.common.type.SnapshotContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.io.UncheckedIOException;
+import java.io.Writer;
+import java.util.Map;
+
+/**
+ * Class to store snapshot data of Materialized view source tables.
+ * The data represents the state of the source tables when the view was created/last rebuilt.
+ */
+public class MaterializationSnapshot {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MaterializationSnapshot.class);
+
+ public static MaterializationSnapshot fromJson(String jsonString) {
+ try {
+ return new ObjectMapper().readValue(jsonString, MaterializationSnapshot.class);
+ } catch (JsonProcessingException e) {
+ LOG.warn(String.format("Unable to parse string as json '%s'. Falling back to parse it as a ValidTxnWriteIdList",
+ jsonString), e);
+ return new MaterializationSnapshot(jsonString);
+ }
+ }
+
+ // Snapshot of native ACID tables.
+ private String validTxnList;
+ // Snapshot of non-native ACID and insert-only transactional tables. Key is the fully qualified name of the table.
+ // Value is the unique id of the snapshot provided by the table's storage HiveStorageHandler.
+ private Map<String, SnapshotContext> tableSnapshots;
+
+ /**
+ * Constructor for json serialization
+ */
+ private MaterializationSnapshot() {
+ }
+
+ public MaterializationSnapshot(String validTxnList) {
+ this.validTxnList = validTxnList;
+ this.tableSnapshots = null;
+ }
+
+ public MaterializationSnapshot(Map<String, SnapshotContext> tableSnapshots) {
+ this.validTxnList = null;
+ this.tableSnapshots = tableSnapshots;
+ }
+
+ /**
+ * Returns the json representation of this object.
+ * @return {@link String} containing a json.
+ */
+ public String asJsonString() {
+ try (Writer out = new StringWriter()) {
+ new ObjectMapper().writeValue(out, this);
+ return out.toString();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Unable to convert " + this + " to json", e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "MaterializationSnapshot{" +
+ "validTxnList='" + validTxnList + '\'' +
+ ", tableSnapshots=" + tableSnapshots +
+ '}';
+ }
+
+ /**
+ * Gets {@link ValidTxnWriteIdList} of the MV source tables when the MV was created/last rebuilt
+ * @return {@link String} representation of {@link ValidTxnWriteIdList} or null if all source tables are non-native.
+ */
+ public String getValidTxnList() {
+ return validTxnList;
+ }
+
+ /**
+ * Gets the snapshotIds of the MV source tables when the MV was created/last rebuilt
+ * @return {@link Map} of snapshotIds where the key is the fully qualified name of the table and the
+ * values is the {@link String} representation of snapshotId or null if all tables are native ACID
+ */
+ public Map<String, SnapshotContext> getTableSnapshots() {
+ return tableSnapshots;
+ }
+}
diff --git a/storage-api/src/java/org/apache/hadoop/hive/common/type/SnapshotContext.java b/storage-api/src/java/org/apache/hadoop/hive/common/type/SnapshotContext.java
new file mode 100644
index 00000000000..40daba97e6b
--- /dev/null
+++ b/storage-api/src/java/org/apache/hadoop/hive/common/type/SnapshotContext.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.common.type;
+
+import java.util.Objects;
+
+/**
+ * Represents a table snapshot.
+ * This is used by transferring relevant snapshot info using HiveStorageHandler API.
+ * Currently, it wraps only Iceberg snapshotId to support decision-making whether to use
+ * materialized view on Iceberg tables.
+ */
+public class SnapshotContext {
+ private long snapshotId;
+
+ /**
+ * Constructor for json serialization
+ */
+ private SnapshotContext() {
+ }
+
+ public SnapshotContext(long snapshotId) {
+ this.snapshotId = snapshotId;
+ }
+
+ /**
+ * Unique identifier of this snapshot context and the underlying snapshot.
+ * @return long snapshotId.
+ */
+ public long getSnapshotId() {
+ return snapshotId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SnapshotContext that = (SnapshotContext) o;
+ return snapshotId == that.snapshotId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(snapshotId);
+ }
+}