HIVE-28817: Rebuilding a materialized view stored in Iceberg fails when schema has varchar column (Krisztian Kasa, reviewed by Laszlo Bodor)

Author: kasakrisz

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/MetastoreLock.java

@@ -21,8 +21,6 @@
 
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.Executors;
@@ -33,6 +31,8 @@
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.LockRequestBuilder;
+import org.apache.hadoop.hive.metastore.api.DataOperationType;
 import org.apache.hadoop.hive.metastore.api.LockComponent;
 import org.apache.hadoop.hive.metastore.api.LockLevel;
 import org.apache.hadoop.hive.metastore.api.LockRequest;
@@ -42,12 +42,12 @@
 import org.apache.hadoop.hive.metastore.api.ShowLocksRequest;
 import org.apache.hadoop.hive.metastore.api.ShowLocksResponse;
 import org.apache.hadoop.hive.metastore.api.ShowLocksResponseElement;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.iceberg.ClientPool;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.iceberg.util.Tasks;
 import org.apache.thrift.TException;
@@ -91,6 +91,7 @@ public class MetastoreLock implements HiveLock {
   private final long lockHeartbeatIntervalTime;
   private final ScheduledExecutorService exitingScheduledExecutorService;
   private final String agentInfo;
+  private final Configuration conf;
 
   private Optional<Long> hmsLockId = Optional.empty();
   private ReentrantLock jvmLock = null;
@@ -102,6 +103,7 @@ public MetastoreLock(Configuration conf, ClientPool<IMetaStoreClient, TException
     this.fullName = catalogName + "." + databaseName + "." + tableName;
     this.databaseName = databaseName;
     this.tableName = tableName;
+    this.conf = conf;
 
     this.lockAcquireTimeout =
         conf.getLong(HIVE_ACQUIRE_LOCK_TIMEOUT_MS, HIVE_ACQUIRE_LOCK_TIMEOUT_MS_DEFAULT);
@@ -268,21 +270,18 @@ private long acquireLock() throws LockException {
   private LockInfo createLock() throws LockException {
     LockInfo lockInfo = new LockInfo();
 
-    String hostName;
-    try {
-      hostName = InetAddress.getLocalHost().getHostName();
-    } catch (UnknownHostException uhe) {
-      throw new LockException(uhe, "Error generating host name");
-    }
-
     LockComponent lockComponent =
             new LockComponent(LockType.EXCL_WRITE, LockLevel.TABLE, databaseName);
+    lockComponent.setOperationType(DataOperationType.NO_TXN);
     lockComponent.setTablename(tableName);
-    LockRequest lockRequest =
-            new LockRequest(
-                    Lists.newArrayList(lockComponent),
-                    HiveHadoopUtil.currentUser(),
-                    hostName);
+
+    // An open ACID transaction might exist when the SQL statement involves both native and Iceberg tables.
+    // Use it's txn id.
+    LockRequest lockRequest = new LockRequestBuilder(null)
+            .setTransactionId(conf.getLong(hive_metastoreConstants.TXN_ID, 0L))
+            .setUser(HiveHadoopUtil.currentUser())
+            .addLockComponent(lockComponent)
+            .build();
 
     // Only works in Hive 2 or later.
     if (HiveVersion.min(HiveVersion.HIVE_2)) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java

@@ -414,6 +414,7 @@ public void preAlterTable(org.apache.hadoop.hive.metastore.api.Table hmsTable, E
     }
 
     try {
+      conf.setLong(hive_metastoreConstants.TXN_ID, IcebergAcidUtil.getTxnId());
       commitLock.lock();
       doPreAlterTable(hmsTable, context);
     } catch (Exception e) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergOutputCommitter.java

@@ -43,6 +43,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.ql.Context.Operation;
 import org.apache.hadoop.hive.ql.Context.RewritePolicy;
 import org.apache.hadoop.hive.ql.exec.Utilities;
@@ -456,6 +457,7 @@ private void commitTable(FileIO io, ExecutorService executor, OutputTable output
 
     for (JobContext jobContext : jobContexts) {
       JobConf conf = jobContext.getJobConf();
+      conf.setLong(hive_metastoreConstants.TXN_ID, IcebergAcidUtil.getTxnId());
       table = Optional.ofNullable(table).orElse(Catalogs.loadTable(conf, catalogProperties));
       branchName = conf.get(InputFormatConfig.OUTPUT_TABLE_SNAPSHOT_REF);
       snapshotId = getSnapshotId(outputTable.table, branchName);

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergAcidUtil.java

@@ -28,7 +28,9 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.ql.io.PositionDeleteInfo;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
 import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
+import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.PartitionKey;
@@ -391,4 +393,17 @@ public T build() {
     }
   }
 
+  static long getTxnId() {
+    SessionState sessionState = SessionState.get();
+    if (sessionState == null) {
+      return 0L;
+    }
+
+    HiveTxnManager txnManager = sessionState.getTxnMgr();
+    if (txnManager == null) {
+      return 0L;
+    }
+
+    return txnManager.getCurrentTxnId();
+  }
 }

iceberg/iceberg-handler/src/test/queries/positive/mv_iceberg_orc8.q

@@ -0,0 +1,23 @@
+-- MV source table has varchar column.
+-- SORT_QUERY_RESULTS
+
+set hive.explain.user=false;
+set hive.support.concurrency=true;
+set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
+set iceberg.mr.schema.auto.conversion=true;
+
+create table t1 (a int, b varchar(256), c char(100), d int not null) stored as orc tblproperties ('transactional'='true');
+
+insert into t1 values (1, 'Alfred', 'Alfred', 1);
+
+create materialized view mat1 stored by iceberg stored as orc tblproperties ('format-version'='2') as
+select b, c, d, sum(a) from t1 group by b, c, d;
+
+insert into t1 values (4, 'Jane', 'Jane', 4);
+
+explain
+alter materialized view mat1 rebuild;
+
+alter materialized view mat1 rebuild;
+
+select * from mat1;