Skip to content

Small improvements in Hive and Iceberg #9605

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
findepi merged 3 commits into from
Oct 12, 2021
Merged

Small improvements in Hive and Iceberg #9605

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9ee4c31
Unlock HMS lock on timeout of failure
findepi Oct 11, 2021
0f710a3
Delete temp dir explicitly
findepi Oct 11, 2021
eca22db
Use correct identity when talking to metastore in Iceberg
findepi Oct 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions ...rc/main/java/io/trino/plugin/hive/metastore/thrift/FailureAwareThriftMetastoreClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,6 +377,13 @@ public LockResponse checkLock(long lockId)
return runWithHandle(() -> delegate.checkLock(lockId));
}

@Override
public void unlock(long lockId)
throws TException
{
runWithHandle(() -> delegate.unlock(lockId));
}

@Override
public String getValidWriteIds(List<String> tableList, long currentTransactionId)
throws TException
Expand Down
38 changes: 34 additions & 4 deletions ...n/trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftHiveMetastore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1676,13 +1676,12 @@ private void acquireSharedLock(DataOperationType operation, boolean isDynamicPar
}
}));

long lockId = response.getLockid();
long waitStart = nanoTime();
while (response.getState() == LockState.WAITING) {
long lockId = response.getLockid();

if (Duration.nanosSince(waitStart).compareTo(maxWaitForLock) > 0) {
// timed out
throw new TrinoException(HIVE_TABLE_LOCK_NOT_ACQUIRED, format("Timed out waiting for lock %d in hive transaction %s for query %s", lockId, transactionId, queryId));
throw unlockSuppressing(identity, lockId, new TrinoException(HIVE_TABLE_LOCK_NOT_ACQUIRED, format("Timed out waiting for lock %d in hive transaction %s for query %s", lockId, transactionId, queryId)));
}

log.debug("Waiting for lock %d in hive transaction %s for query %s", lockId, transactionId, queryId);
Expand All @@ -1697,7 +1696,7 @@ private void acquireSharedLock(DataOperationType operation, boolean isDynamicPar
}

if (response.getState() != LockState.ACQUIRED) {
throw new TrinoException(HIVE_TABLE_LOCK_NOT_ACQUIRED, "Could not acquire lock. Lock in state " + response.getState());
throw unlockSuppressing(identity, lockId, new TrinoException(HIVE_TABLE_LOCK_NOT_ACQUIRED, "Could not acquire lock. Lock in state " + response.getState()));
}
}
catch (TException e) {
Expand All @@ -1708,6 +1707,37 @@ private void acquireSharedLock(DataOperationType operation, boolean isDynamicPar
}
}

private <T extends Exception> T unlockSuppressing(HiveIdentity identity, long lockId, T exception)
{
try {
unlockTableLock(identity, lockId);
}
catch (RuntimeException e) {
exception.addSuppressed(e);
}
return exception;
}

private void unlockTableLock(HiveIdentity identity, long lockId)
{
try {
retry()
.stopOn(NoSuchTxnException.class, NoSuchLockException.class, TxnAbortedException.class, MetaException.class)
.run("unlock", stats.getUnlock().wrap(() -> {
try (ThriftMetastoreClient metastoreClient = createMetastoreClient(identity)) {
metastoreClient.unlock(lockId);
}
return null;
}));
}
catch (TException e) {
throw new TrinoException(HIVE_METASTORE_ERROR, e);
}
catch (Exception e) {
throw propagate(e);
}
}

private static LockComponent createLockComponentForOperation(SchemaTableName table, DataOperationType operation, boolean isDynamicPartitionWrite, Optional<String> partitionName)
{
requireNonNull(table, "table is null");
Expand Down
8 changes: 8 additions & 0 deletions ...o-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftHiveMetastoreClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
import org.apache.hadoop.hive.metastore.api.TableStatsRequest;
import org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore;
import org.apache.hadoop.hive.metastore.api.TxnToWriteId;
import org.apache.hadoop.hive.metastore.api.UnlockRequest;
import org.apache.hadoop.hive.metastore.txn.TxnUtils;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
Expand Down Expand Up @@ -503,6 +504,13 @@ public LockResponse checkLock(long lockId)
return client.check_lock(new CheckLockRequest(lockId));
}

@Override
public void unlock(long lockId)
throws TException
{
client.unlock(new UnlockRequest(lockId));
}

@Override
public String getValidWriteIds(List<String> tableList, long currentTransactionId)
throws TException
Expand Down
3 changes: 3 additions & 0 deletions ...trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftMetastoreClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,9 @@ LockResponse acquireLock(LockRequest lockRequest)
LockResponse checkLock(long lockId)
throws TException;

void unlock(long lockId)
throws TException;

String getValidWriteIds(List<String> tableList, long currentTransactionId)
throws TException;

Expand Down
8 changes: 8 additions & 0 deletions .../trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftMetastoreStats.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ public class ThriftMetastoreStats
private final ThriftMetastoreApiStats rollbackTransaction = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats acquireLock = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats checkLock = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats unlock = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats validWriteIds = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats allocateWriteId = new ThriftMetastoreApiStats();
private final ThriftMetastoreApiStats updateTableWriteId = new ThriftMetastoreApiStats();
Expand Down Expand Up @@ -353,6 +354,13 @@ public ThriftMetastoreApiStats getCheckLock()
return checkLock;
}

@Managed
@Nested
public ThriftMetastoreApiStats getUnlock()
{
return unlock;
}

@Managed
@Nested
public ThriftMetastoreApiStats getValidWriteIds()
Expand Down
7 changes: 7 additions & 0 deletions ...o-hive/src/test/java/io/trino/plugin/hive/metastore/thrift/MockThriftMetastoreClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -478,6 +478,13 @@ public LockResponse checkLock(long lockId)
throw new UnsupportedOperationException();
}

@Override
public void unlock(long lockId)
throws TException
{
throw new UnsupportedOperationException();
}

@Override
public String getValidWriteIds(List<String> tableList, long currentTransactionId)
{
Expand Down
7 changes: 2 additions & 5 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/HiveTableOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import io.trino.spi.TrinoException;
import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.connector.TableNotFoundException;
import io.trino.spi.security.ConnectorIdentity;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
import org.apache.hadoop.mapred.FileInputFormat;
Expand Down Expand Up @@ -221,13 +220,11 @@ public void commit(@Nullable TableMetadata base, TableMetadata metadata)
}

PrincipalPrivileges privileges = buildInitialPrivilegeSet(table.getOwner());
ConnectorIdentity identity = ConnectorIdentity.ofUser(table.getOwner());
HiveIdentity context = new HiveIdentity(identity);
if (base == null) {
metastore.createTable(context, table, privileges);
metastore.createTable(identity, table, privileges);
}
else {
metastore.replaceTable(context, database, tableName, table, privileges);
metastore.replaceTable(identity, database, tableName, table, privileges);
}

shouldRefresh = true;
Expand Down
7 changes: 4 additions & 3 deletions plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ public class TestIcebergV2
{
private HiveMetastore metastore;
private HdfsEnvironment hdfsEnvironment;
private java.nio.file.Path tempDir;
private File metastoreDir;

@Override
Expand All @@ -75,8 +76,8 @@ protected QueryRunner createQueryRunner()
HdfsConfiguration configuration = new HiveHdfsConfiguration(new HdfsConfigurationInitializer(config), ImmutableSet.of());
hdfsEnvironment = new HdfsEnvironment(configuration, config, new NoHdfsAuthentication());

File tempDir = Files.createTempDirectory("test_iceberg_v2").toFile();
metastoreDir = new File(tempDir, "iceberg_data");
tempDir = Files.createTempDirectory("test_iceberg_v2");
metastoreDir = tempDir.resolve("iceberg_data").toFile();
metastore = createTestingFileHiveMetastore(metastoreDir);

return createIcebergQueryRunner(ImmutableMap.of(), ImmutableMap.of(), ImmutableList.of(NATION), Optional.of(metastoreDir));
Expand All @@ -86,7 +87,7 @@ protected QueryRunner createQueryRunner()
public void tearDown()
throws IOException
{
deleteRecursively(metastoreDir.getParentFile().toPath(), ALLOW_INSECURE);
deleteRecursively(tempDir, ALLOW_INSECURE);
}

@Test
Expand Down