You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/08/26 22:53:58 UTC
[incubator-iceberg] branch master updated: Support create and
replace transactions in Catalog (#362)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new d7c7d0f Support create and replace transactions in Catalog (#362)
d7c7d0f is described below
commit d7c7d0fbb28d711a4e4320c596cfdf4ce1caf93d
Author: Anton Okolnychyi <ao...@apple.com>
AuthorDate: Mon Aug 26 23:53:53 2019 +0100
Support create and replace transactions in Catalog (#362)
---
.../java/org/apache/iceberg/catalog/Catalog.java | 141 ++++++++++++
build.gradle | 2 +
.../org/apache/iceberg/BaseMetastoreCatalog.java | 43 ++++
.../iceberg/BaseMetastoreTableOperations.java | 30 +++
.../java/org/apache/iceberg/BaseTransaction.java | 43 ++--
.../main/java/org/apache/iceberg/Transactions.java | 11 +-
.../apache/iceberg/hive/HiveTableOperations.java | 12 +-
.../iceberg/hive/HiveCreateReplaceTableTest.java | 256 +++++++++++++++++++++
.../org/apache/iceberg/hive/HiveMetastoreTest.java | 61 +++++
.../org/apache/iceberg/hive/HiveTableBaseTest.java | 38 +--
10 files changed, 575 insertions(+), 62 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/catalog/Catalog.java b/api/src/main/java/org/apache/iceberg/catalog/Catalog.java
index 2461080..8a6cd74 100644
--- a/api/src/main/java/org/apache/iceberg/catalog/Catalog.java
+++ b/api/src/main/java/org/apache/iceberg/catalog/Catalog.java
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -98,6 +99,146 @@ public interface Catalog {
}
/**
+ * Start a transaction to create a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param location a location for the table; leave null if unspecified
+ * @param properties a string map of table properties
+ * @return a {@link Transaction} to create the table
+ * @throws AlreadyExistsException if the table already exists
+ */
+ Transaction newCreateTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ String location,
+ Map<String, String> properties);
+
+ /**
+ * Start a transaction to create a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param properties a string map of table properties
+ * @return a {@link Transaction} to create the table
+ * @throws AlreadyExistsException if the table already exists
+ */
+ default Transaction newCreateTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ Map<String, String> properties) {
+ return newCreateTableTransaction(identifier, schema, spec, null, properties);
+ }
+
+ /**
+ * Start a transaction to create a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @return a {@link Transaction} to create the table
+ * @throws AlreadyExistsException if the table already exists
+ */
+ default Transaction newCreateTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec) {
+ return newCreateTableTransaction(identifier, schema, spec, null, null);
+ }
+
+ /**
+ * Start a transaction to create a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @return a {@link Transaction} to create the table
+ * @throws AlreadyExistsException if the table already exists
+ */
+ default Transaction newCreateTableTransaction(
+ TableIdentifier identifier,
+ Schema schema) {
+ return newCreateTableTransaction(identifier, schema, PartitionSpec.unpartitioned(), null, null);
+ }
+
+ /**
+ * Start a transaction to replace a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param location a location for the table; leave null if unspecified
+ * @param properties a string map of table properties
+ * @param orCreate whether to create the table if not exists
+ * @return a {@link Transaction} to replace the table
+ * @throws NoSuchTableException if the table doesn't exist and orCreate is false
+ */
+ Transaction newReplaceTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ String location,
+ Map<String, String> properties,
+ boolean orCreate);
+
+ /**
+ * Start a transaction to replace a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param properties a string map of table properties
+ * @param orCreate whether to create the table if not exists
+ * @return a {@link Transaction} to replace the table
+ * @throws NoSuchTableException if the table doesn't exist and orCreate is false
+ */
+ default Transaction newReplaceTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ Map<String, String> properties,
+ boolean orCreate) {
+ return newReplaceTableTransaction(identifier, schema, spec, null, properties, orCreate);
+ }
+
+ /**
+ * Start a transaction to replace a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param spec a partition spec
+ * @param orCreate whether to create the table if not exists
+ * @return a {@link Transaction} to replace the table
+ * @throws NoSuchTableException if the table doesn't exist and orCreate is false
+ */
+ default Transaction newReplaceTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ boolean orCreate) {
+ return newReplaceTableTransaction(identifier, schema, spec, null, null, orCreate);
+ }
+
+ /**
+ * Start a transaction to replace a table.
+ *
+ * @param identifier a table identifier
+ * @param schema a schema
+ * @param orCreate whether to create the table if not exists
+ * @return a {@link Transaction} to replace the table
+ * @throws NoSuchTableException if the table doesn't exist and orCreate is false
+ */
+ default Transaction newReplaceTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ boolean orCreate) {
+ return newReplaceTableTransaction(identifier, schema, PartitionSpec.unpartitioned(), null, null, orCreate);
+ }
+
+ /**
* Check whether table exists.
*
* @param identifier a table identifier
diff --git a/build.gradle b/build.gradle
index a66970d..534f6a5 100644
--- a/build.gradle
+++ b/build.gradle
@@ -306,6 +306,8 @@ project(':iceberg-hive') {
exclude group: 'org.apache.avro', module: 'avro'
exclude group: 'org.slf4j', module: 'slf4j-log4j12'
}
+
+ testCompile project(path: ':iceberg-core', configuration: 'testArtifacts')
}
}
diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
index 60864d9..01e9cb6 100644
--- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
@@ -91,6 +91,49 @@ public abstract class BaseMetastoreCatalog implements Catalog {
}
@Override
+ public Transaction newCreateTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ String location,
+ Map<String, String> properties) {
+
+ TableOperations ops = newTableOps(identifier);
+ if (ops.current() != null) {
+ throw new AlreadyExistsException("Table already exists: " + identifier);
+ }
+
+ String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
+ Map<String, String> tableProperties = properties != null ? properties : Maps.newHashMap();
+ TableMetadata metadata = TableMetadata.newTableMetadata(ops, schema, spec, baseLocation, tableProperties);
+ return Transactions.createTableTransaction(ops, metadata);
+ }
+
+ @Override
+ public Transaction newReplaceTableTransaction(
+ TableIdentifier identifier,
+ Schema schema,
+ PartitionSpec spec,
+ String location,
+ Map<String, String> properties,
+ boolean orCreate) {
+
+ TableOperations ops = newTableOps(identifier);
+ if (!orCreate && ops.current() == null) {
+ throw new NoSuchTableException("No such table: " + identifier);
+ }
+
+ String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
+ Map<String, String> tableProperties = properties != null ? properties : Maps.newHashMap();
+ TableMetadata metadata = TableMetadata.newTableMetadata(ops, schema, spec, baseLocation, tableProperties);
+ if (orCreate) {
+ return Transactions.createOrReplaceTableTransaction(ops, metadata);
+ } else {
+ return Transactions.replaceTableTransaction(ops, metadata);
+ }
+ }
+
+ @Override
public Table loadTable(TableIdentifier identifier) {
TableOperations ops = newTableOps(identifier);
if (ops.current() == null) {
diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
index 55f3305..78bc1d4 100644
--- a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
+++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.io.OutputFile;
@@ -66,6 +67,35 @@ public abstract class BaseMetastoreTableOperations implements TableOperations {
return version;
}
+ @Override
+ public TableMetadata refresh() {
+ try {
+ doRefresh();
+ } catch (NoSuchTableException e) {
+ LOG.warn("Could not find the table during refresh, setting current metadata to null", e);
+ currentMetadata = null;
+ currentMetadataLocation = null;
+ version = -1;
+ shouldRefresh = false;
+ throw e;
+ }
+ return current();
+ }
+
+ protected void doRefresh() {
+ throw new UnsupportedOperationException("Not implemented: doRefresh");
+ }
+
+ @Override
+ public void commit(TableMetadata base, TableMetadata metadata) {
+ doCommit(base, metadata);
+ requestRefresh();
+ }
+
+ protected void doCommit(TableMetadata base, TableMetadata metadata) {
+ throw new UnsupportedOperationException("Not implemented: doCommit");
+ }
+
protected void requestRefresh() {
this.shouldRefresh = true;
}
diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
index 3306270..1ecb652 100644
--- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java
+++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java
@@ -30,8 +30,10 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.util.PropertyUtil;
import org.apache.iceberg.util.Tasks;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,9 +50,10 @@ import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFA
class BaseTransaction implements Transaction {
private static final Logger LOG = LoggerFactory.getLogger(BaseTransaction.class);
- private enum TransactionType {
+ enum TransactionType {
CREATE_TABLE,
REPLACE_TABLE,
+ CREATE_OR_REPLACE_TABLE,
SIMPLE
}
@@ -66,20 +69,14 @@ class BaseTransaction implements Transaction {
private TableMetadata lastBase;
private TableMetadata current;
- BaseTransaction(TableOperations ops, TableMetadata start) {
+ BaseTransaction(TableOperations ops, TransactionType type, TableMetadata start) {
this.ops = ops;
this.transactionTable = new TransactionTable();
this.transactionOps = new TransactionTableOperations();
this.updates = Lists.newArrayList();
this.intermediateSnapshotIds = Sets.newHashSet();
this.base = ops.current();
- if (base == null && start != null) {
- this.type = TransactionType.CREATE_TABLE;
- } else if (base != null && start != base) {
- this.type = TransactionType.REPLACE_TABLE;
- } else {
- this.type = TransactionType.SIMPLE;
- }
+ this.type = type;
this.lastBase = null;
this.current = start;
}
@@ -201,7 +198,11 @@ class BaseTransaction implements Transaction {
break;
case REPLACE_TABLE:
- commitReplaceTransaction();
+ commitReplaceTransaction(false);
+ break;
+
+ case CREATE_OR_REPLACE_TABLE:
+ commitReplaceTransaction(true);
break;
case SIMPLE:
@@ -241,23 +242,33 @@ class BaseTransaction implements Transaction {
}
}
- private void commitReplaceTransaction() {
+ private void commitReplaceTransaction(boolean orCreate) {
// fix up the snapshot log, which should not contain intermediate snapshots
TableMetadata replaceMetadata = current.removeSnapshotLogEntries(intermediateSnapshotIds);
+ Map<String, String> props = base != null ? base.properties() : replaceMetadata.properties();
try {
Tasks.foreach(ops)
- .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
+ .retry(PropertyUtil.propertyAsInt(props, COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
.exponentialBackoff(
- base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
- base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
- base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
+ PropertyUtil.propertyAsInt(props, COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
+ PropertyUtil.propertyAsInt(props, COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
+ PropertyUtil.propertyAsInt(props, COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
2.0 /* exponential */)
.onlyRetryOn(CommitFailedException.class)
.run(underlyingOps -> {
+
+ try {
+ underlyingOps.refresh();
+ } catch (NoSuchTableException e) {
+ if (!orCreate) {
+ throw e;
+ }
+ }
+
// because this is a replace table, it will always completely replace the table
// metadata. even if it was just updated.
- if (base != underlyingOps.refresh()) {
+ if (base != underlyingOps.current()) {
this.base = underlyingOps.current(); // just refreshed
}
diff --git a/core/src/main/java/org/apache/iceberg/Transactions.java b/core/src/main/java/org/apache/iceberg/Transactions.java
index c7708c9..07093ff 100644
--- a/core/src/main/java/org/apache/iceberg/Transactions.java
+++ b/core/src/main/java/org/apache/iceberg/Transactions.java
@@ -20,21 +20,26 @@
package org.apache.iceberg;
import com.google.common.base.Preconditions;
+import org.apache.iceberg.BaseTransaction.TransactionType;
public final class Transactions {
private Transactions() {}
+ public static Transaction createOrReplaceTableTransaction(TableOperations ops, TableMetadata start) {
+ return new BaseTransaction(ops, TransactionType.CREATE_OR_REPLACE_TABLE, start);
+ }
+
public static Transaction replaceTableTransaction(TableOperations ops, TableMetadata start) {
- return new BaseTransaction(ops, start);
+ return new BaseTransaction(ops, TransactionType.REPLACE_TABLE, start);
}
public static Transaction createTableTransaction(TableOperations ops, TableMetadata start) {
Preconditions.checkArgument(ops.current() == null,
"Cannot start create table transaction: table already exists");
- return new BaseTransaction(ops, start);
+ return new BaseTransaction(ops, TransactionType.CREATE_TABLE, start);
}
public static Transaction newTransaction(TableOperations ops) {
- return new BaseTransaction(ops, ops.refresh());
+ return new BaseTransaction(ops, TransactionType.SIMPLE, ops.refresh());
}
}
diff --git a/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java b/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
index 310a9c0..1cb4edc 100644
--- a/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
+++ b/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
@@ -46,6 +46,7 @@ import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.hadoop.HadoopFileIO;
@@ -85,7 +86,7 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
}
@Override
- public TableMetadata refresh() {
+ protected void doRefresh() {
String metadataLocation = null;
try {
final Table table = metaClients.run(client -> client.getTable(database, tableName));
@@ -118,12 +119,10 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
}
refreshFromMetadataLocation(metadataLocation);
-
- return current();
}
@Override
- public void commit(TableMetadata base, TableMetadata metadata) {
+ protected void doCommit(TableMetadata base, TableMetadata metadata) {
// if the metadata is already out of date, reject it
if (base != current()) {
throw new CommitFailedException("Cannot commit: stale table metadata for %s.%s", database, tableName);
@@ -184,6 +183,9 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
});
}
threw = false;
+ } catch (org.apache.hadoop.hive.metastore.api.AlreadyExistsException e) {
+ throw new AlreadyExistsException("Table already exists: %s.%s", database, tableName);
+
} catch (TException | UnknownHostException e) {
throw new RuntimeException(String.format("Metastore operation failed for %s.%s", database, tableName), e);
@@ -198,8 +200,6 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
}
unlock(lockId);
}
-
- requestRefresh();
}
private void setParameters(String newMetadataLocation, Table tbl) {
diff --git a/hive/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java b/hive/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
new file mode 100644
index 0000000..fd5aa74
--- /dev/null
+++ b/hive/src/test/java/org/apache/iceberg/hive/HiveCreateReplaceTableTest.java
@@ -0,0 +1,256 @@
+/*
+ * 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.iceberg.hive;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import static org.apache.iceberg.PartitionSpec.builderFor;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+public class HiveCreateReplaceTableTest extends HiveMetastoreTest {
+
+ private static final String TABLE_NAME = "tbl";
+ private static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of(DB_NAME, TABLE_NAME);
+ private static final Schema SCHEMA = new Schema(
+ required(3, "id", Types.IntegerType.get()),
+ required(4, "data", Types.StringType.get())
+ );
+ private static final PartitionSpec SPEC = builderFor(SCHEMA)
+ .identity("id")
+ .build();
+
+ @Rule
+ public TemporaryFolder temp = new TemporaryFolder();
+
+ private String tableLocation;
+
+ @Before
+ public void createTableLocation() throws IOException {
+ tableLocation = temp.newFolder("hive-").getPath();
+ }
+
+ @After
+ public void cleanup() {
+ catalog.dropTable(TABLE_IDENTIFIER);
+ }
+
+ @Test
+ public void testCreateTableTxn() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newCreateTableTransaction(
+ TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+
+ // verify the table is still not visible before the transaction is committed
+ Assert.assertFalse(catalog.tableExists(TABLE_IDENTIFIER));
+
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Table props should match", "value", table.properties().get("prop"));
+ }
+
+ @Test
+ public void testCreateTableTxnTableCreatedConcurrently() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newCreateTableTransaction(
+ TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+
+ // create the table concurrently
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC);
+ Assert.assertTrue("Table should be created", catalog.tableExists(TABLE_IDENTIFIER));
+
+ AssertHelpers.assertThrows(
+ "Create table txn should fail",
+ AlreadyExistsException.class,
+ "Table already exists: hivedb.tbl",
+ txn::commitTransaction);
+ }
+
+ @Test
+ public void testCreateTableTxnTableAlreadyExists() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ // create a table before starting a transaction
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC);
+ Assert.assertTrue("Table should be created", catalog.tableExists(TABLE_IDENTIFIER));
+
+ AssertHelpers.assertThrows(
+ "Should not be possible to start a new create table txn",
+ AlreadyExistsException.class,
+ "Table already exists: hivedb.tbl",
+ () -> catalog.newCreateTableTransaction(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap()));
+ }
+
+ @Test
+ public void testReplaceTableTxn() {
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+ Assert.assertTrue("Table should exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, false);
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Partition spec should match", PartitionSpec.unpartitioned(), table.spec());
+ }
+
+ @Test
+ public void testReplaceTableTxnTableNotExists() {
+ AssertHelpers.assertThrows(
+ "Should not be possible to start a new replace table txn",
+ NoSuchTableException.class,
+ "No such table: hivedb.tbl",
+ () -> catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, SPEC, false));
+ }
+
+ @Test
+ public void testReplaceTableTxnTableDeletedConcurrently() {
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+ Assert.assertTrue("Table should exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, SPEC, false);
+
+ catalog.dropTable(TABLE_IDENTIFIER);
+
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+
+ AssertHelpers.assertThrows(
+ "Replace table txn should fail",
+ NoSuchTableException.class,
+ "No such table: hivedb.tbl",
+ txn::commitTransaction);
+ }
+
+ @Test
+ public void testReplaceTableTxnTableModifiedConcurrently() {
+ Table table = catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+ Assert.assertTrue("Table should exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, SPEC, false);
+
+ // update the table concurrently
+ table.updateProperties()
+ .set("another-prop", "another-value")
+ .commit();
+
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+ txn.commitTransaction();
+
+ // the replace should still succeed
+ table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertNull("Table props should be updated", table.properties().get("another-prop"));
+ Assert.assertEquals("Table props should match", "value", table.properties().get("prop"));
+ }
+
+ @Test
+ public void testCreateOrReplaceTableTxnTableNotExists() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, SPEC, true);
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Table props should match", "value", table.properties().get("prop"));
+ }
+
+ @Test
+ public void testCreateOrReplaceTableTxnTableExists() {
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC, tableLocation, Maps.newHashMap());
+ Assert.assertTrue("Table should exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(TABLE_IDENTIFIER, SCHEMA, true);
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Partition spec should match", PartitionSpec.unpartitioned(), table.spec());
+ }
+
+ @Test
+ public void testCreateOrReplaceTableTxnTableDeletedConcurrently() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC);
+ Assert.assertTrue("Table should be created", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(
+ TABLE_IDENTIFIER, SCHEMA, PartitionSpec.unpartitioned(), tableLocation, Maps.newHashMap(), true);
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+
+ // drop the table concurrently
+ catalog.dropTable(TABLE_IDENTIFIER);
+
+ // expect the transaction to succeed anyway
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Table props should match", "value", table.properties().get("prop"));
+ }
+
+ @Test
+ public void testCreateOrReplaceTableTxnTableCreatedConcurrently() {
+ Assert.assertFalse("Table should not exist", catalog.tableExists(TABLE_IDENTIFIER));
+
+ Transaction txn = catalog.newReplaceTableTransaction(
+ TABLE_IDENTIFIER, SCHEMA, PartitionSpec.unpartitioned(), tableLocation, Maps.newHashMap(), true);
+ txn.updateProperties()
+ .set("prop", "value")
+ .commit();
+
+ // create the table concurrently
+ catalog.createTable(TABLE_IDENTIFIER, SCHEMA, SPEC);
+ Assert.assertTrue("Table should be created", catalog.tableExists(TABLE_IDENTIFIER));
+
+ // expect the transaction to succeed anyway
+ txn.commitTransaction();
+
+ Table table = catalog.loadTable(TABLE_IDENTIFIER);
+ Assert.assertEquals("Partition spec should match", PartitionSpec.unpartitioned(), table.spec());
+ Assert.assertEquals("Table props should match", "value", table.properties().get("prop"));
+ }
+}
diff --git a/hive/src/test/java/org/apache/iceberg/hive/HiveMetastoreTest.java b/hive/src/test/java/org/apache/iceberg/hive/HiveMetastoreTest.java
new file mode 100644
index 0000000..8d08aff
--- /dev/null
+++ b/hive/src/test/java/org/apache/iceberg/hive/HiveMetastoreTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.iceberg.hive;
+
+import java.util.HashMap;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public abstract class HiveMetastoreTest {
+
+ protected static final String DB_NAME = "hivedb";
+
+ protected static HiveMetaStoreClient metastoreClient;
+ protected static HiveCatalog catalog;
+ protected static HiveConf hiveConf;
+ protected static TestHiveMetastore metastore;
+
+ @BeforeClass
+ public static void startMetastore() throws Exception {
+ HiveMetastoreTest.metastore = new TestHiveMetastore();
+ metastore.start();
+ HiveMetastoreTest.hiveConf = metastore.hiveConf();
+ HiveMetastoreTest.metastoreClient = new HiveMetaStoreClient(hiveConf);
+ String dbPath = metastore.getDatabasePath(DB_NAME);
+ Database db = new Database(DB_NAME, "description", dbPath, new HashMap<>());
+ metastoreClient.createDatabase(db);
+ HiveMetastoreTest.catalog = new HiveCatalog(hiveConf);
+ }
+
+ @AfterClass
+ public static void stopMetastore() {
+ catalog.close();
+ HiveMetastoreTest.catalog = null;
+
+ metastoreClient.close();
+ HiveMetastoreTest.metastoreClient = null;
+
+ metastore.stop();
+ HiveMetastoreTest.metastore = null;
+ }
+}
diff --git a/hive/src/test/java/org/apache/iceberg/hive/HiveTableBaseTest.java b/hive/src/test/java/org/apache/iceberg/hive/HiveTableBaseTest.java
index 1776610..76f2192 100644
--- a/hive/src/test/java/org/apache/iceberg/hive/HiveTableBaseTest.java
+++ b/hive/src/test/java/org/apache/iceberg/hive/HiveTableBaseTest.java
@@ -22,22 +22,16 @@ package org.apache.iceberg.hive;
import java.io.File;
import java.nio.file.Paths;
import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.TableMetadataParser;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.types.Types;
import org.junit.After;
-import org.junit.AfterClass;
import org.junit.Before;
-import org.junit.BeforeClass;
import static org.apache.iceberg.PartitionSpec.builderFor;
import static org.apache.iceberg.TableMetadataParser.getFileExtension;
@@ -45,9 +39,8 @@ import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
-public class HiveTableBaseTest {
+public class HiveTableBaseTest extends HiveMetastoreTest {
- protected static final String DB_NAME = "hivedb";
static final String TABLE_NAME = "tbl";
static final TableIdentifier TABLE_IDENTIFIER = TableIdentifier.of(DB_NAME, TABLE_NAME);
@@ -60,35 +53,6 @@ public class HiveTableBaseTest {
private static final PartitionSpec partitionSpec = builderFor(schema).identity("id").build();
- private static TestHiveMetastore metastore;
-
- protected static HiveMetaStoreClient metastoreClient;
- protected static HiveCatalog catalog;
- protected static HiveConf hiveConf;
-
- @BeforeClass
- public static void startMetastore() throws Exception {
- HiveTableBaseTest.metastore = new TestHiveMetastore();
- metastore.start();
- HiveTableBaseTest.hiveConf = metastore.hiveConf();
- HiveTableBaseTest.metastoreClient = new HiveMetaStoreClient(hiveConf);
- String dbPath = metastore.getDatabasePath(DB_NAME);
- Database db = new Database(DB_NAME, "description", dbPath, new HashMap<>());
- metastoreClient.createDatabase(db);
- HiveTableBaseTest.catalog = new HiveCatalog(hiveConf);
- }
-
- @AfterClass
- public static void stopMetastore() {
- HiveTableBaseTest.catalog.close();
-
- metastoreClient.close();
- HiveTableBaseTest.metastoreClient = null;
-
- metastore.stop();
- HiveTableBaseTest.metastore = null;
- }
-
private Path tableLocation;
@Before