You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2020/06/17 21:58:43 UTC
[GitHub] [hive] miklosgergely opened a new pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
miklosgergely opened a new pull request #1142:
URL: https://github.com/apache/hive/pull/1142
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442101534
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -308,110 +255,6 @@ public FetchTask getFetchTask() {
return driverContext.getFetchTask();
}
- /**
- * Acquire read and write locks needed by the statement. The list of objects to be locked are
- * obtained from the inputs and outputs populated by the compiler. Locking strategy depends on
- * HiveTxnManager and HiveLockManager configured
- *
- * This method also records the list of valid transactions. This must be done after any
- * transactions have been opened.
- * @throws CommandProcessorException
- **/
- private void acquireLocks() throws CommandProcessorException {
- PerfLogger perfLogger = SessionState.getPerfLogger();
- perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
-
- if(!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
- /*non acid txn managers don't support txns but fwd lock requests to lock managers
- acid txn manager requires all locks to be associated with a txn so if we
- end up here w/o an open txn it's because we are processing something like "use <database>
- which by definition needs no locks*/
- return;
- }
- try {
- String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
-
- // Set the table write id in all of the acid file sinks
- if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
- List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
- //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
- //so this makes (file name -> data) mapping stable
- acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) ->
- fsd1.getDirName().compareTo(fsd2.getDirName()));
- for (FileSinkDesc desc : acidSinks) {
- TableDesc tableInfo = desc.getTableInfo();
- final TableName tn = HiveTableName.ofNullable(tableInfo.getTableName());
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- desc.setTableWriteId(writeId);
-
- /**
- * it's possible to have > 1 FileSink writing to the same table/partition
- * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
- * Insert ... Select ... Union All Select ... using
- * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
- */
- desc.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
- String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
- if(desc.getInsertOverwrite() && desc.getDirName().toString().contains(unionAllSubdir) &&
- desc.isFullAcidTable()) {
- throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
- " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
- }
- }
- }
-
- if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
- // Allocate write ID for the table being analyzed.
- Table t = driverContext.getPlan().getAcidAnalyzeTable().getTable();
- driverContext.getTxnManager().getTableWriteId(t.getDbName(), t.getTableName());
- }
-
-
- DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
- boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
- if (hasAcidDdl) {
- String fqTableName = acidDdlDesc.getFullTableName();
- final TableName tn = HiveTableName.ofNullableWithNoDefault(fqTableName);
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- acidDdlDesc.setWriteId(writeId);
- }
-
- /*It's imperative that {@code acquireLocks()} is called for all commands so that
- HiveTxnManager can transition its state machine correctly*/
- driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
- final List<HiveLock> locks = context.getHiveLocks();
- LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
- ((locks == null) ? 0 : locks.size()));
- // This check is for controlling the correctness of the current state
- if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
- !driverContext.isValidTxnListsGenerated()) {
- throw new IllegalStateException(
- "Need to record valid WriteID list but there is no valid TxnID list (" +
- JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
- ", queryId:" + driverContext.getPlan().getQueryId() + ")");
- }
-
- if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
- validTxnManager.recordValidWriteIds();
- }
-
- } catch (Exception e) {
- String errorMessage;
- if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
- errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
- driverState.toString(), e.getMessage());
- CONSOLE.printInfo(errorMessage);
- } else {
- errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
- CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
- }
- throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
- e);
- } finally {
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- }
- }
-
public void releaseLocksAndCommitOrRollback(boolean commit) throws LockException {
releaseLocksAndCommitOrRollback(commit, driverContext.getTxnManager());
}
Review comment:
What about removing Driver.releaseLocksAndCommitOrRollback altogether, and call driverTxnHandler.releaseLocksAndCommitOrRollback(commit, txnManager) immediately?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442505140
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1112,15 +865,15 @@ public void destroy() {
boolean isTxnOpen = driverContext != null
&& driverContext.getTxnManager() != null
&& driverContext.getTxnManager().isTxnOpen();
- if (!hiveLocks.isEmpty() || isTxnOpen) {
+ if (!driverTxnHandler.hasHiveLock() || isTxnOpen) {
Review comment:
I agree, I'll modify this.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1058,15 +811,15 @@ private int closeInProcess(boolean destroyed) {
releaseResStream();
releaseContext();
if (destroyed) {
- if (!hiveLocks.isEmpty()) {
+ if (!driverTxnHandler.hasHiveLock()) {
Review comment:
I agree, I'll modify this.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -565,9 +368,9 @@ public void lockAndRespond() throws CommandProcessorException {
"No previously compiled query for driver - queryId=" + driverContext.getQueryState().getQueryId());
}
- if (requiresLock()) {
Review comment:
I agree, I'll modify this.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -565,9 +368,9 @@ public void lockAndRespond() throws CommandProcessorException {
"No previously compiled query for driver - queryId=" + driverContext.getQueryState().getQueryId());
}
- if (requiresLock()) {
Review comment:
I agree, I'll modify it.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1058,15 +811,15 @@ private int closeInProcess(boolean destroyed) {
releaseResStream();
releaseContext();
if (destroyed) {
- if (!hiveLocks.isEmpty()) {
+ if (!driverTxnHandler.hasHiveLock()) {
Review comment:
I agree, I'll modify this.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1112,15 +865,15 @@ public void destroy() {
boolean isTxnOpen = driverContext != null
&& driverContext.getTxnManager() != null
&& driverContext.getTxnManager().isTxnOpen();
- if (!hiveLocks.isEmpty() || isTxnOpen) {
+ if (!driverTxnHandler.hasHiveLock() || isTxnOpen) {
Review comment:
I agree, I'll modify this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442105791
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1112,15 +865,15 @@ public void destroy() {
boolean isTxnOpen = driverContext != null
&& driverContext.getTxnManager() != null
&& driverContext.getTxnManager().isTxnOpen();
- if (!hiveLocks.isEmpty() || isTxnOpen) {
+ if (!driverTxnHandler.hasHiveLock() || isTxnOpen) {
Review comment:
See comment above: DriverTxnHanlder.destroy?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442701763
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -988,7 +741,7 @@ private void releaseContext() {
}
context.clear(deleteResultDir);
if (context.getHiveLocks() != null) {
- hiveLocks.addAll(context.getHiveLocks());
+ driverTxnHandler.addHiveLocksFromContext();
Review comment:
Ok
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442508297
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -308,110 +255,6 @@ public FetchTask getFetchTask() {
return driverContext.getFetchTask();
}
- /**
- * Acquire read and write locks needed by the statement. The list of objects to be locked are
- * obtained from the inputs and outputs populated by the compiler. Locking strategy depends on
- * HiveTxnManager and HiveLockManager configured
- *
- * This method also records the list of valid transactions. This must be done after any
- * transactions have been opened.
- * @throws CommandProcessorException
- **/
- private void acquireLocks() throws CommandProcessorException {
- PerfLogger perfLogger = SessionState.getPerfLogger();
- perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
-
- if(!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
- /*non acid txn managers don't support txns but fwd lock requests to lock managers
- acid txn manager requires all locks to be associated with a txn so if we
- end up here w/o an open txn it's because we are processing something like "use <database>
- which by definition needs no locks*/
- return;
- }
- try {
- String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
-
- // Set the table write id in all of the acid file sinks
- if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
- List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
- //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
- //so this makes (file name -> data) mapping stable
- acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) ->
- fsd1.getDirName().compareTo(fsd2.getDirName()));
- for (FileSinkDesc desc : acidSinks) {
- TableDesc tableInfo = desc.getTableInfo();
- final TableName tn = HiveTableName.ofNullable(tableInfo.getTableName());
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- desc.setTableWriteId(writeId);
-
- /**
- * it's possible to have > 1 FileSink writing to the same table/partition
- * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
- * Insert ... Select ... Union All Select ... using
- * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
- */
- desc.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
- String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
- if(desc.getInsertOverwrite() && desc.getDirName().toString().contains(unionAllSubdir) &&
- desc.isFullAcidTable()) {
- throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
- " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
- }
- }
- }
-
- if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
- // Allocate write ID for the table being analyzed.
- Table t = driverContext.getPlan().getAcidAnalyzeTable().getTable();
- driverContext.getTxnManager().getTableWriteId(t.getDbName(), t.getTableName());
- }
-
-
- DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
- boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
- if (hasAcidDdl) {
- String fqTableName = acidDdlDesc.getFullTableName();
- final TableName tn = HiveTableName.ofNullableWithNoDefault(fqTableName);
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- acidDdlDesc.setWriteId(writeId);
- }
-
- /*It's imperative that {@code acquireLocks()} is called for all commands so that
- HiveTxnManager can transition its state machine correctly*/
- driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
- final List<HiveLock> locks = context.getHiveLocks();
- LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
- ((locks == null) ? 0 : locks.size()));
- // This check is for controlling the correctness of the current state
- if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
- !driverContext.isValidTxnListsGenerated()) {
- throw new IllegalStateException(
- "Need to record valid WriteID list but there is no valid TxnID list (" +
- JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
- ", queryId:" + driverContext.getPlan().getQueryId() + ")");
- }
-
- if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
- validTxnManager.recordValidWriteIds();
- }
-
- } catch (Exception e) {
- String errorMessage;
- if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
- errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
- driverState.toString(), e.getMessage());
- CONSOLE.printInfo(errorMessage);
- } else {
- errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
- CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
- }
- throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
- e);
- } finally {
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- }
- }
-
public void releaseLocksAndCommitOrRollback(boolean commit) throws LockException {
releaseLocksAndCommitOrRollback(commit, driverContext.getTxnManager());
}
Review comment:
I'm planning to remove both releaseLocksAndCommitOrRollback functions from Driver, but unfortunately they are both used by some external callers. In a future patch I'll try to remove those.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442508297
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -308,110 +255,6 @@ public FetchTask getFetchTask() {
return driverContext.getFetchTask();
}
- /**
- * Acquire read and write locks needed by the statement. The list of objects to be locked are
- * obtained from the inputs and outputs populated by the compiler. Locking strategy depends on
- * HiveTxnManager and HiveLockManager configured
- *
- * This method also records the list of valid transactions. This must be done after any
- * transactions have been opened.
- * @throws CommandProcessorException
- **/
- private void acquireLocks() throws CommandProcessorException {
- PerfLogger perfLogger = SessionState.getPerfLogger();
- perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
-
- if(!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
- /*non acid txn managers don't support txns but fwd lock requests to lock managers
- acid txn manager requires all locks to be associated with a txn so if we
- end up here w/o an open txn it's because we are processing something like "use <database>
- which by definition needs no locks*/
- return;
- }
- try {
- String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
-
- // Set the table write id in all of the acid file sinks
- if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
- List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
- //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
- //so this makes (file name -> data) mapping stable
- acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) ->
- fsd1.getDirName().compareTo(fsd2.getDirName()));
- for (FileSinkDesc desc : acidSinks) {
- TableDesc tableInfo = desc.getTableInfo();
- final TableName tn = HiveTableName.ofNullable(tableInfo.getTableName());
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- desc.setTableWriteId(writeId);
-
- /**
- * it's possible to have > 1 FileSink writing to the same table/partition
- * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
- * Insert ... Select ... Union All Select ... using
- * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
- */
- desc.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
- String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
- if(desc.getInsertOverwrite() && desc.getDirName().toString().contains(unionAllSubdir) &&
- desc.isFullAcidTable()) {
- throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
- " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
- }
- }
- }
-
- if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
- // Allocate write ID for the table being analyzed.
- Table t = driverContext.getPlan().getAcidAnalyzeTable().getTable();
- driverContext.getTxnManager().getTableWriteId(t.getDbName(), t.getTableName());
- }
-
-
- DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
- boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
- if (hasAcidDdl) {
- String fqTableName = acidDdlDesc.getFullTableName();
- final TableName tn = HiveTableName.ofNullableWithNoDefault(fqTableName);
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- acidDdlDesc.setWriteId(writeId);
- }
-
- /*It's imperative that {@code acquireLocks()} is called for all commands so that
- HiveTxnManager can transition its state machine correctly*/
- driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
- final List<HiveLock> locks = context.getHiveLocks();
- LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
- ((locks == null) ? 0 : locks.size()));
- // This check is for controlling the correctness of the current state
- if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
- !driverContext.isValidTxnListsGenerated()) {
- throw new IllegalStateException(
- "Need to record valid WriteID list but there is no valid TxnID list (" +
- JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
- ", queryId:" + driverContext.getPlan().getQueryId() + ")");
- }
-
- if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
- validTxnManager.recordValidWriteIds();
- }
-
- } catch (Exception e) {
- String errorMessage;
- if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
- errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
- driverState.toString(), e.getMessage());
- CONSOLE.printInfo(errorMessage);
- } else {
- errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
- CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
- }
- throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
- e);
- } finally {
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- }
- }
-
public void releaseLocksAndCommitOrRollback(boolean commit) throws LockException {
releaseLocksAndCommitOrRollback(commit, driverContext.getTxnManager());
}
Review comment:
I'm planning to remove both releaseLocksAndCommitOrRollback functions from Driver, but unfortunately they are both used by some external callers. In a future patch I'll try to remove those. I've removed all the references to them within the Driver class for now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442516604
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/DriverTxnHandler.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.ql;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.hive.common.JavaUtils;
+import org.apache.hadoop.hive.common.TableName;
+import org.apache.hadoop.hive.common.ValidTxnList;
+import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
+import org.apache.hadoop.hive.conf.Constants;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.ql.ddl.DDLDesc.DDLDescWithWriteId;
+import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator;
+import org.apache.hadoop.hive.ql.exec.ConditionalTask;
+import org.apache.hadoop.hive.ql.exec.Task;
+import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
+import org.apache.hadoop.hive.ql.lockmgr.LockException;
+import org.apache.hadoop.hive.ql.log.PerfLogger;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.HiveTableName;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
+import org.apache.hadoop.hive.ql.plan.HiveOperation;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorException;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hive.common.util.ShutdownHookManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ */
+public class DriverTxnHandler {
+ private static final String CLASS_NAME = Driver.class.getName();
+ private static final Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
+ private static final LogHelper CONSOLE = new LogHelper(LOG);
+ private static final int SHUTDOWN_HOOK_PRIORITY = 0;
+
+ private final DriverContext driverContext;
+ private final DriverState driverState;
+ private final ValidTxnManager validTxnManager;
+
+ private final List<HiveLock> hiveLocks = new ArrayList<HiveLock>();
+
+ private Context context;
+
+ public DriverTxnHandler(DriverContext driverContext, DriverState driverState, ValidTxnManager validTxnManager) {
+ this.driverContext = driverContext;
+ this.driverState = driverState;
+ this.validTxnManager = validTxnManager;
+ }
+
+ public void createTxnManager() throws CommandProcessorException {
+ try {
+ // Initialize the transaction manager. This must be done before analyze is called.
+ HiveTxnManager queryTxnManager = (driverContext.getInitTxnManager() != null) ?
+ driverContext.getInitTxnManager() : SessionState.get().initTxnMgr(driverContext.getConf());
+
+ if (queryTxnManager instanceof Configurable) {
+ ((Configurable) queryTxnManager).setConf(driverContext.getConf());
+ }
+ driverContext.setTxnManager(queryTxnManager);
+ driverContext.getQueryState().setTxnManager(queryTxnManager);
+
+ // In case when user Ctrl-C twice to kill Hive CLI JVM, we want to release locks
+ // if compile is being called multiple times, clear the old shutdownhook
+ ShutdownHookManager.removeShutdownHook(driverContext.getShutdownRunner());
+ Runnable shutdownRunner = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ releaseLocksAndCommitOrRollback(false, driverContext.getTxnManager());
+ } catch (LockException e) {
+ LOG.warn("Exception when releasing locks in ShutdownHook for Driver: " +
+ e.getMessage());
+ }
+ }
+ };
+ ShutdownHookManager.addShutdownHook(shutdownRunner, SHUTDOWN_HOOK_PRIORITY);
+ driverContext.setShutdownRunner(shutdownRunner);
+ } catch (LockException e) {
+ ErrorMsg error = ErrorMsg.getErrorMsg(e.getMessage());
+ String errorMessage = "FAILED: " + e.getClass().getSimpleName() + " [Error " + error.getErrorCode() + "]:";
+
+ CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
+ throw DriverUtils.createProcessorException(driverContext, error.getErrorCode(), errorMessage, error.getSQLState(),
+ e);
+ }
+ }
+
+ public void setContext(Context context) {
+ this.context = context;
+ }
+
+ public boolean requiresLock() {
+ if (!DriverUtils.checkConcurrency(driverContext)) {
+ LOG.info("Concurrency mode is disabled, not creating a lock manager");
+ return false;
+ }
+
+ // Lock operations themselves don't require the lock.
+ if (isExplicitLockOperation()) {
+ return false;
+ }
+
+ if (!HiveConf.getBoolVar(driverContext.getConf(), ConfVars.HIVE_LOCK_MAPRED_ONLY)) {
+ return true;
+ }
+
+ if (driverContext.getConf().get(Constants.HIVE_QUERY_EXCLUSIVE_LOCK) != null) {
+ return true;
+ }
+
+ Queue<Task<?>> tasks = new LinkedList<Task<?>>();
+ tasks.addAll(driverContext.getPlan().getRootTasks());
+ while (tasks.peek() != null) {
+ Task<?> task = tasks.remove();
+ if (task.requireLock()) {
+ return true;
+ }
+
+ if (task instanceof ConditionalTask) {
+ tasks.addAll(((ConditionalTask)task).getListTasks());
+ }
+
+ if (task.getChildTasks() != null) {
+ tasks.addAll(task.getChildTasks());
+ }
+ // does not add back up task here, because back up task should be the same type of the original task.
+ }
+
+ return false;
+ }
+
+ private boolean isExplicitLockOperation() {
+ HiveOperation currentOpt = driverContext.getPlan().getOperation();
+ if (currentOpt != null) {
+ switch (currentOpt) {
+ case LOCKDB:
+ case UNLOCKDB:
+ case LOCKTABLE:
+ case UNLOCKTABLE:
+ return true;
+ default:
+ return false;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Acquire read and write locks needed by the statement. The list of objects to be locked are obtained from the inputs
+ * and outputs populated by the compiler. Locking strategy depends on HiveTxnManager and HiveLockManager configured.
+ *
+ * This method also records the list of valid transactions. This must be done after any transactions have been opened.
+ */
+ public void acquireLocks() throws CommandProcessorException {
+ PerfLogger perfLogger = SessionState.getPerfLogger();
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
+
+ if (!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
+ /* non acid txn managers don't support txns but fwd lock requests to lock managers
+ acid txn manager requires all locks to be associated with a txn so if we end up here w/o an open txn
+ it's because we are processing something like "use <database> which by definition needs no locks */
+ return;
+ }
+
+ try {
+ setWriteIdForAcidFileSinks();
+ allocateWriteIdForAcidAnalyzeTable();
+ boolean hasAcidDdl = setWriteIdForAcidDdl();
+ acquireLocksInternal();
+
+ if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
+ validTxnManager.recordValidWriteIds();
+ }
+ } catch (Exception e) {
+ String errorMessage;
+ if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
+ errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
+ driverState.toString(), e.getMessage());
+ CONSOLE.printInfo(errorMessage);
+ } else {
+ errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
+ CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
+ }
+ throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
+ e);
+ } finally {
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
+ }
+ }
+
+ private void setWriteIdForAcidFileSinks() throws SemanticException, LockException {
+ if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
+ List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
+ //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
+ //so this makes (file name -> data) mapping stable
+ acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) -> fsd1.getDirName().compareTo(fsd2.getDirName()));
+ for (FileSinkDesc acidSink : acidSinks) {
+ TableDesc tableInfo = acidSink.getTableInfo();
+ TableName tableName = HiveTableName.of(tableInfo.getTableName());
+ long writeId = driverContext.getTxnManager().getTableWriteId(tableName.getDb(), tableName.getTable());
+ acidSink.setTableWriteId(writeId);
+
+ /**
+ * it's possible to have > 1 FileSink writing to the same table/partition
+ * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
+ * Insert ... Select ... Union All Select ... using
+ * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
+ */
+ acidSink.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
+ String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
+ if (acidSink.getInsertOverwrite() && acidSink.getDirName().toString().contains(unionAllSubdir) &&
+ acidSink.isFullAcidTable()) {
+ throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
+ " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
+ }
+ }
+ }
+ }
+
+ private void allocateWriteIdForAcidAnalyzeTable() throws LockException {
+ if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
+ Table table = driverContext.getPlan().getAcidAnalyzeTable().getTable();
+ driverContext.getTxnManager().getTableWriteId(table.getDbName(), table.getTableName());
+ }
+ }
+
+ private boolean setWriteIdForAcidDdl() throws SemanticException, LockException {
+ DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
+ boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
+ if (hasAcidDdl) {
+ String fqTableName = acidDdlDesc.getFullTableName();
+ TableName tableName = HiveTableName.of(fqTableName);
+ long writeId = driverContext.getTxnManager().getTableWriteId(tableName.getDb(), tableName.getTable());
+ acidDdlDesc.setWriteId(writeId);
+ }
+ return hasAcidDdl;
+ }
+
+ private void acquireLocksInternal() throws CommandProcessorException, LockException {
+ /* It's imperative that {@code acquireLocks()} is called for all commands so that
+ HiveTxnManager can transition its state machine correctly */
+ String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
+ driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
+ List<HiveLock> locks = context.getHiveLocks();
+ LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
+ ((locks == null) ? 0 : locks.size()));
+ // This check is for controlling the correctness of the current state
+ if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
+ !driverContext.isValidTxnListsGenerated()) {
+ throw new IllegalStateException("Need to record valid WriteID list but there is no valid TxnID list (" +
+ JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
+ ", queryId: " + driverContext.getPlan().getQueryId() + ")");
+ }
+ }
+
+ public void addHiveLocksFromContext() {
+ hiveLocks.addAll(context.getHiveLocks());
+ }
+
+ public boolean hasHiveLock() {
+ return hiveLocks.isEmpty();
+ }
Review comment:
I agree, that HiveLocks should be handled together. Currently MoveTask also needs to access them, as it wants to remove somme locks, so to have the hiveLocks only locally in DriverTxnHandler should somehow solve that. I suggest to solve this issue in a separate jira, as this one is only for moving code to a cleaner structure.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442701113
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -308,110 +255,6 @@ public FetchTask getFetchTask() {
return driverContext.getFetchTask();
}
- /**
- * Acquire read and write locks needed by the statement. The list of objects to be locked are
- * obtained from the inputs and outputs populated by the compiler. Locking strategy depends on
- * HiveTxnManager and HiveLockManager configured
- *
- * This method also records the list of valid transactions. This must be done after any
- * transactions have been opened.
- * @throws CommandProcessorException
- **/
- private void acquireLocks() throws CommandProcessorException {
- PerfLogger perfLogger = SessionState.getPerfLogger();
- perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
-
- if(!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
- /*non acid txn managers don't support txns but fwd lock requests to lock managers
- acid txn manager requires all locks to be associated with a txn so if we
- end up here w/o an open txn it's because we are processing something like "use <database>
- which by definition needs no locks*/
- return;
- }
- try {
- String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
-
- // Set the table write id in all of the acid file sinks
- if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
- List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
- //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
- //so this makes (file name -> data) mapping stable
- acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) ->
- fsd1.getDirName().compareTo(fsd2.getDirName()));
- for (FileSinkDesc desc : acidSinks) {
- TableDesc tableInfo = desc.getTableInfo();
- final TableName tn = HiveTableName.ofNullable(tableInfo.getTableName());
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- desc.setTableWriteId(writeId);
-
- /**
- * it's possible to have > 1 FileSink writing to the same table/partition
- * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
- * Insert ... Select ... Union All Select ... using
- * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
- */
- desc.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
- String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
- if(desc.getInsertOverwrite() && desc.getDirName().toString().contains(unionAllSubdir) &&
- desc.isFullAcidTable()) {
- throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
- " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
- }
- }
- }
-
- if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
- // Allocate write ID for the table being analyzed.
- Table t = driverContext.getPlan().getAcidAnalyzeTable().getTable();
- driverContext.getTxnManager().getTableWriteId(t.getDbName(), t.getTableName());
- }
-
-
- DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
- boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
- if (hasAcidDdl) {
- String fqTableName = acidDdlDesc.getFullTableName();
- final TableName tn = HiveTableName.ofNullableWithNoDefault(fqTableName);
- long writeId = driverContext.getTxnManager().getTableWriteId(tn.getDb(), tn.getTable());
- acidDdlDesc.setWriteId(writeId);
- }
-
- /*It's imperative that {@code acquireLocks()} is called for all commands so that
- HiveTxnManager can transition its state machine correctly*/
- driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
- final List<HiveLock> locks = context.getHiveLocks();
- LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
- ((locks == null) ? 0 : locks.size()));
- // This check is for controlling the correctness of the current state
- if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
- !driverContext.isValidTxnListsGenerated()) {
- throw new IllegalStateException(
- "Need to record valid WriteID list but there is no valid TxnID list (" +
- JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
- ", queryId:" + driverContext.getPlan().getQueryId() + ")");
- }
-
- if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
- validTxnManager.recordValidWriteIds();
- }
-
- } catch (Exception e) {
- String errorMessage;
- if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
- errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
- driverState.toString(), e.getMessage());
- CONSOLE.printInfo(errorMessage);
- } else {
- errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
- CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
- }
- throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
- e);
- } finally {
- perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
- }
- }
-
public void releaseLocksAndCommitOrRollback(boolean commit) throws LockException {
releaseLocksAndCommitOrRollback(commit, driverContext.getTxnManager());
}
Review comment:
Sounds good
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442100824
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -565,9 +368,9 @@ public void lockAndRespond() throws CommandProcessorException {
"No previously compiled query for driver - queryId=" + driverContext.getQueryState().getQueryId());
}
- if (requiresLock()) {
Review comment:
What about driverTxnHandler.acquireLocksIfNeeded, and pushing the logic to the TxnHandler?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on pull request #1142:
URL: https://github.com/apache/hive/pull/1142#issuecomment-646312272
I totally agree about merging ValidTxnManager into DriverTxnHandler, I was already considering it, I'm glad that you see the same thing. I'd prefer to do it in a separate patch, so it would be easier to follow what have happened. Once this one is in, I'll merge them, is that ok?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] miklosgergely commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
miklosgergely commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442515235
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -988,7 +741,7 @@ private void releaseContext() {
}
context.clear(deleteResultDir);
if (context.getHiveLocks() != null) {
- hiveLocks.addAll(context.getHiveLocks());
+ driverTxnHandler.addHiveLocksFromContext();
Review comment:
The main problem here is that we have some HiveLocks in the hiveLocks variable (previously in Driver, after this patch in DriverTxnHandler), and also some HiveLocks in the Context. The two should be unified I believe, probably in Context. I think it's a good goal, but I'd rather not put it into this patch, as this is only moving code around. Let's do this in another jira, do you agree?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on pull request #1142:
URL: https://github.com/apache/hive/pull/1142#issuecomment-645912312
I like this cleanup stuff. We should separate put Txn related code from the Driver. Maybe worth to consider later to merge with ValidTxnManager? But even without that it is a worthy change
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442105092
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -988,7 +741,7 @@ private void releaseContext() {
}
context.clear(deleteResultDir);
if (context.getHiveLocks() != null) {
- hiveLocks.addAll(context.getHiveLocks());
+ driverTxnHandler.addHiveLocksFromContext();
Review comment:
Do we need this magic? DriverTxnHandler already has a Context.
If I understand correctly this is just clearing the context at hand, should be the responsibility of the DriverTxnHandler.... or even better it can have a DriverTxnHandlerContext...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442105554
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/Driver.java
##########
@@ -1058,15 +811,15 @@ private int closeInProcess(boolean destroyed) {
releaseResStream();
releaseContext();
if (destroyed) {
- if (!hiveLocks.isEmpty()) {
+ if (!driverTxnHandler.hasHiveLock()) {
Review comment:
Again, maybe just a driverTxnHandler.release?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pvary commented on a change in pull request #1142: HIVE-23718 Extract transaction handling from Driver (Miklos Gergely)
Posted by GitBox <gi...@apache.org>.
pvary commented on a change in pull request #1142:
URL: https://github.com/apache/hive/pull/1142#discussion_r442106620
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/DriverTxnHandler.java
##########
@@ -0,0 +1,336 @@
+/*
+ * 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.ql;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.hive.common.JavaUtils;
+import org.apache.hadoop.hive.common.TableName;
+import org.apache.hadoop.hive.common.ValidTxnList;
+import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
+import org.apache.hadoop.hive.conf.Constants;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.ql.ddl.DDLDesc.DDLDescWithWriteId;
+import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator;
+import org.apache.hadoop.hive.ql.exec.ConditionalTask;
+import org.apache.hadoop.hive.ql.exec.Task;
+import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
+import org.apache.hadoop.hive.ql.lockmgr.LockException;
+import org.apache.hadoop.hive.ql.log.PerfLogger;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.parse.HiveTableName;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
+import org.apache.hadoop.hive.ql.plan.HiveOperation;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorException;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hive.common.util.ShutdownHookManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ */
+public class DriverTxnHandler {
+ private static final String CLASS_NAME = Driver.class.getName();
+ private static final Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
+ private static final LogHelper CONSOLE = new LogHelper(LOG);
+ private static final int SHUTDOWN_HOOK_PRIORITY = 0;
+
+ private final DriverContext driverContext;
+ private final DriverState driverState;
+ private final ValidTxnManager validTxnManager;
+
+ private final List<HiveLock> hiveLocks = new ArrayList<HiveLock>();
+
+ private Context context;
+
+ public DriverTxnHandler(DriverContext driverContext, DriverState driverState, ValidTxnManager validTxnManager) {
+ this.driverContext = driverContext;
+ this.driverState = driverState;
+ this.validTxnManager = validTxnManager;
+ }
+
+ public void createTxnManager() throws CommandProcessorException {
+ try {
+ // Initialize the transaction manager. This must be done before analyze is called.
+ HiveTxnManager queryTxnManager = (driverContext.getInitTxnManager() != null) ?
+ driverContext.getInitTxnManager() : SessionState.get().initTxnMgr(driverContext.getConf());
+
+ if (queryTxnManager instanceof Configurable) {
+ ((Configurable) queryTxnManager).setConf(driverContext.getConf());
+ }
+ driverContext.setTxnManager(queryTxnManager);
+ driverContext.getQueryState().setTxnManager(queryTxnManager);
+
+ // In case when user Ctrl-C twice to kill Hive CLI JVM, we want to release locks
+ // if compile is being called multiple times, clear the old shutdownhook
+ ShutdownHookManager.removeShutdownHook(driverContext.getShutdownRunner());
+ Runnable shutdownRunner = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ releaseLocksAndCommitOrRollback(false, driverContext.getTxnManager());
+ } catch (LockException e) {
+ LOG.warn("Exception when releasing locks in ShutdownHook for Driver: " +
+ e.getMessage());
+ }
+ }
+ };
+ ShutdownHookManager.addShutdownHook(shutdownRunner, SHUTDOWN_HOOK_PRIORITY);
+ driverContext.setShutdownRunner(shutdownRunner);
+ } catch (LockException e) {
+ ErrorMsg error = ErrorMsg.getErrorMsg(e.getMessage());
+ String errorMessage = "FAILED: " + e.getClass().getSimpleName() + " [Error " + error.getErrorCode() + "]:";
+
+ CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
+ throw DriverUtils.createProcessorException(driverContext, error.getErrorCode(), errorMessage, error.getSQLState(),
+ e);
+ }
+ }
+
+ public void setContext(Context context) {
+ this.context = context;
+ }
+
+ public boolean requiresLock() {
+ if (!DriverUtils.checkConcurrency(driverContext)) {
+ LOG.info("Concurrency mode is disabled, not creating a lock manager");
+ return false;
+ }
+
+ // Lock operations themselves don't require the lock.
+ if (isExplicitLockOperation()) {
+ return false;
+ }
+
+ if (!HiveConf.getBoolVar(driverContext.getConf(), ConfVars.HIVE_LOCK_MAPRED_ONLY)) {
+ return true;
+ }
+
+ if (driverContext.getConf().get(Constants.HIVE_QUERY_EXCLUSIVE_LOCK) != null) {
+ return true;
+ }
+
+ Queue<Task<?>> tasks = new LinkedList<Task<?>>();
+ tasks.addAll(driverContext.getPlan().getRootTasks());
+ while (tasks.peek() != null) {
+ Task<?> task = tasks.remove();
+ if (task.requireLock()) {
+ return true;
+ }
+
+ if (task instanceof ConditionalTask) {
+ tasks.addAll(((ConditionalTask)task).getListTasks());
+ }
+
+ if (task.getChildTasks() != null) {
+ tasks.addAll(task.getChildTasks());
+ }
+ // does not add back up task here, because back up task should be the same type of the original task.
+ }
+
+ return false;
+ }
+
+ private boolean isExplicitLockOperation() {
+ HiveOperation currentOpt = driverContext.getPlan().getOperation();
+ if (currentOpt != null) {
+ switch (currentOpt) {
+ case LOCKDB:
+ case UNLOCKDB:
+ case LOCKTABLE:
+ case UNLOCKTABLE:
+ return true;
+ default:
+ return false;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Acquire read and write locks needed by the statement. The list of objects to be locked are obtained from the inputs
+ * and outputs populated by the compiler. Locking strategy depends on HiveTxnManager and HiveLockManager configured.
+ *
+ * This method also records the list of valid transactions. This must be done after any transactions have been opened.
+ */
+ public void acquireLocks() throws CommandProcessorException {
+ PerfLogger perfLogger = SessionState.getPerfLogger();
+ perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
+
+ if (!driverContext.getTxnManager().isTxnOpen() && driverContext.getTxnManager().supportsAcid()) {
+ /* non acid txn managers don't support txns but fwd lock requests to lock managers
+ acid txn manager requires all locks to be associated with a txn so if we end up here w/o an open txn
+ it's because we are processing something like "use <database> which by definition needs no locks */
+ return;
+ }
+
+ try {
+ setWriteIdForAcidFileSinks();
+ allocateWriteIdForAcidAnalyzeTable();
+ boolean hasAcidDdl = setWriteIdForAcidDdl();
+ acquireLocksInternal();
+
+ if (driverContext.getPlan().hasAcidResourcesInQuery() || hasAcidDdl) {
+ validTxnManager.recordValidWriteIds();
+ }
+ } catch (Exception e) {
+ String errorMessage;
+ if (driverState.isDestroyed() || driverState.isAborted() || driverState.isClosed()) {
+ errorMessage = String.format("Ignore lock acquisition related exception in terminal state (%s): %s",
+ driverState.toString(), e.getMessage());
+ CONSOLE.printInfo(errorMessage);
+ } else {
+ errorMessage = String.format("FAILED: Error in acquiring locks: %s", e.getMessage());
+ CONSOLE.printError(errorMessage, "\n" + StringUtils.stringifyException(e));
+ }
+ throw DriverUtils.createProcessorException(driverContext, 10, errorMessage, ErrorMsg.findSQLState(e.getMessage()),
+ e);
+ } finally {
+ perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.ACQUIRE_READ_WRITE_LOCKS);
+ }
+ }
+
+ private void setWriteIdForAcidFileSinks() throws SemanticException, LockException {
+ if (!driverContext.getPlan().getAcidSinks().isEmpty()) {
+ List<FileSinkDesc> acidSinks = new ArrayList<>(driverContext.getPlan().getAcidSinks());
+ //sorting makes tests easier to write since file names and ROW__IDs depend on statementId
+ //so this makes (file name -> data) mapping stable
+ acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) -> fsd1.getDirName().compareTo(fsd2.getDirName()));
+ for (FileSinkDesc acidSink : acidSinks) {
+ TableDesc tableInfo = acidSink.getTableInfo();
+ TableName tableName = HiveTableName.of(tableInfo.getTableName());
+ long writeId = driverContext.getTxnManager().getTableWriteId(tableName.getDb(), tableName.getTable());
+ acidSink.setTableWriteId(writeId);
+
+ /**
+ * it's possible to have > 1 FileSink writing to the same table/partition
+ * e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes
+ * Insert ... Select ... Union All Select ... using
+ * {@link org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator#UNION_SUDBIR_PREFIX}
+ */
+ acidSink.setStatementId(driverContext.getTxnManager().getStmtIdAndIncrement());
+ String unionAllSubdir = "/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX;
+ if (acidSink.getInsertOverwrite() && acidSink.getDirName().toString().contains(unionAllSubdir) &&
+ acidSink.isFullAcidTable()) {
+ throw new UnsupportedOperationException("QueryId=" + driverContext.getPlan().getQueryId() +
+ " is not supported due to OVERWRITE and UNION ALL. Please use truncate + insert");
+ }
+ }
+ }
+ }
+
+ private void allocateWriteIdForAcidAnalyzeTable() throws LockException {
+ if (driverContext.getPlan().getAcidAnalyzeTable() != null) {
+ Table table = driverContext.getPlan().getAcidAnalyzeTable().getTable();
+ driverContext.getTxnManager().getTableWriteId(table.getDbName(), table.getTableName());
+ }
+ }
+
+ private boolean setWriteIdForAcidDdl() throws SemanticException, LockException {
+ DDLDescWithWriteId acidDdlDesc = driverContext.getPlan().getAcidDdlDesc();
+ boolean hasAcidDdl = acidDdlDesc != null && acidDdlDesc.mayNeedWriteId();
+ if (hasAcidDdl) {
+ String fqTableName = acidDdlDesc.getFullTableName();
+ TableName tableName = HiveTableName.of(fqTableName);
+ long writeId = driverContext.getTxnManager().getTableWriteId(tableName.getDb(), tableName.getTable());
+ acidDdlDesc.setWriteId(writeId);
+ }
+ return hasAcidDdl;
+ }
+
+ private void acquireLocksInternal() throws CommandProcessorException, LockException {
+ /* It's imperative that {@code acquireLocks()} is called for all commands so that
+ HiveTxnManager can transition its state machine correctly */
+ String userFromUGI = DriverUtils.getUserFromUGI(driverContext);
+ driverContext.getTxnManager().acquireLocks(driverContext.getPlan(), context, userFromUGI, driverState);
+ List<HiveLock> locks = context.getHiveLocks();
+ LOG.info("Operation {} obtained {} locks", driverContext.getPlan().getOperation(),
+ ((locks == null) ? 0 : locks.size()));
+ // This check is for controlling the correctness of the current state
+ if (driverContext.getTxnManager().recordSnapshot(driverContext.getPlan()) &&
+ !driverContext.isValidTxnListsGenerated()) {
+ throw new IllegalStateException("Need to record valid WriteID list but there is no valid TxnID list (" +
+ JavaUtils.txnIdToString(driverContext.getTxnManager().getCurrentTxnId()) +
+ ", queryId: " + driverContext.getPlan().getQueryId() + ")");
+ }
+ }
+
+ public void addHiveLocksFromContext() {
+ hiveLocks.addAll(context.getHiveLocks());
+ }
+
+ public boolean hasHiveLock() {
+ return hiveLocks.isEmpty();
+ }
Review comment:
I would prefer keeping the locks local to DriverTxnManager, and its' own context(?), and not storing in the global context?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org