You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by td...@apache.org on 2018/08/06 21:32:46 UTC
phoenix git commit: PHOENIX-4799 Write cells using checkAndMutate to
prevent conflicting changes
Repository: phoenix
Updated Branches:
refs/heads/4.x-HBase-1.2 2541634da -> c9b8d114c
PHOENIX-4799 Write cells using checkAndMutate to prevent conflicting changes
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c9b8d114
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c9b8d114
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c9b8d114
Branch: refs/heads/4.x-HBase-1.2
Commit: c9b8d114c5bf7e4230c11b0714f9ca853d667145
Parents: 2541634
Author: Thomas D'Silva <td...@apache.org>
Authored: Tue Jul 24 10:49:31 2018 -0700
Committer: Thomas D'Silva <td...@apache.org>
Committed: Mon Aug 6 10:53:07 2018 -0700
----------------------------------------------------------------------
.../phoenix/end2end/AlterTableWithViewsIT.java | 2 +-
.../phoenix/end2end/BasePermissionsIT.java | 66 +++---
.../phoenix/end2end/ChangePermissionsIT.java | 4 +-
.../MigrateSystemTablesToSystemNamespaceIT.java | 13 +-
.../end2end/QueryDatabaseMetaDataIT.java | 4 +
.../phoenix/end2end/TableDDLPermissionsIT.java | 9 +-
.../end2end/TenantSpecificTablesDDLIT.java | 2 +
.../org/apache/phoenix/end2end/UpgradeIT.java | 34 +--
.../java/org/apache/phoenix/end2end/ViewIT.java | 235 +++++++++++++++++--
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 2 +
.../phoenix/query/ConnectionQueryServices.java | 14 ++
.../query/ConnectionQueryServicesImpl.java | 170 +++++++-------
.../query/ConnectionlessQueryServicesImpl.java | 20 ++
.../query/DelegateConnectionQueryServices.java | 11 +
.../apache/phoenix/query/QueryConstants.java | 14 ++
.../apache/phoenix/schema/MetaDataClient.java | 95 +++++++-
16 files changed, 523 insertions(+), 172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java
index e39d492..e97a40d 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java
@@ -530,7 +530,7 @@ public class AlterTableWithViewsIT extends SplitSystemCatalogIT {
try {
// should fail because there are two view with different pk columns
- conn.createStatement().execute("ALTER TABLE " + tableName + " ADD VIEW_COL1 DECIMAL PRIMARY KEY, VIEW_COL2 VARCHAR PRIMARY KEY");
+ conn.createStatement().execute("ALTER TABLE " + tableName + " ADD VIEW_COL1 DECIMAL(10,2) PRIMARY KEY, VIEW_COL2 VARCHAR(256) PRIMARY KEY");
fail();
}
catch (SQLException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
index d33d538..88a942e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java
@@ -16,8 +16,29 @@
*/
package org.apache.phoenix.end2end;
-import com.google.common.base.Joiner;
-import com.google.common.base.Throwables;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.security.PrivilegedExceptionAction;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -36,34 +57,13 @@ import org.apache.phoenix.query.BaseTest;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import java.io.IOException;
-import java.lang.reflect.UndeclaredThrowableException;
-import java.security.PrivilegedExceptionAction;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.google.common.base.Joiner;
+import com.google.common.base.Throwables;
@RunWith(Parameterized.class)
public class BasePermissionsIT extends BaseTest {
@@ -73,17 +73,23 @@ public class BasePermissionsIT extends BaseTest {
static String SUPERUSER;
static HBaseTestingUtility testUtil;
- static final Set<String> PHOENIX_SYSTEM_TABLES = new HashSet<>(Arrays.asList(
- "SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS", "SYSTEM.FUNCTION"));
+ static final Set<String> PHOENIX_SYSTEM_TABLES =
+ new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
+ "SYSTEM.FUNCTION", "SYSTEM.MUTEX"));
- static final Set<String> PHOENIX_SYSTEM_TABLES_IDENTIFIERS = new HashSet<>(Arrays.asList(
- "SYSTEM.\"CATALOG\"", "SYSTEM.\"SEQUENCE\"", "SYSTEM.\"STATS\"", "SYSTEM.\"FUNCTION\""));
+ static final Set<String> PHOENIX_SYSTEM_TABLES_IDENTIFIERS =
+ new HashSet<>(Arrays.asList("SYSTEM.\"CATALOG\"", "SYSTEM.\"SEQUENCE\"",
+ "SYSTEM.\"STATS\"", "SYSTEM.\"FUNCTION\"", "SYSTEM.\"MUTEX\""));
static final String SYSTEM_SEQUENCE_IDENTIFIER =
QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\"" + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE+ "\"";
+ static final String SYSTEM_MUTEX_IDENTIFIER =
+ QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\""
+ + PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME + "\"";
+
static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES = new HashSet<>(Arrays.asList(
- "SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS", "SYSTEM:FUNCTION"));
+ "SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS", "SYSTEM:FUNCTION", "SYSTEM:MUTEX"));
// Create Multiple users so that we can use Hadoop UGI to run tasks as various users
// Permissions can be granted or revoke by superusers and admins only
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java
index 3965f69..7f2964d 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java
@@ -57,7 +57,8 @@ public class ChangePermissionsIT extends BasePermissionsIT {
} else {
verifyAllowed(grantPermissions("RX", user, PHOENIX_SYSTEM_TABLES_IDENTIFIERS, false), superUser);
}
- verifyAllowed(grantPermissions("W", user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+ verifyAllowed(grantPermissions("RWX", user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+ verifyAllowed(grantPermissions("RWX", user, SYSTEM_MUTEX_IDENTIFIER, false), superUser);
}
}
@@ -69,6 +70,7 @@ public class ChangePermissionsIT extends BasePermissionsIT {
verifyAllowed(revokePermissions(user, PHOENIX_SYSTEM_TABLES_IDENTIFIERS, false), superUser);
}
verifyAllowed(revokePermissions(user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+ verifyAllowed(revokePermissions(user, SYSTEM_MUTEX_IDENTIFIER, false), superUser);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
index d253f6e..ffac4d6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MigrateSystemTablesToSystemNamespaceIT.java
@@ -54,7 +54,6 @@ import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
import org.junit.After;
-import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -275,16 +274,15 @@ public class MigrateSystemTablesToSystemNamespaceIT extends BaseTest {
private void changeMutexLock(Properties clientProps, boolean acquire) throws SQLException, IOException {
ConnectionQueryServices services = null;
- byte[] mutexRowKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE);
try (Connection conn = DriverManager.getConnection(getJdbcUrl(), clientProps)) {
services = conn.unwrap(PhoenixConnection.class).getQueryServices();
if(acquire) {
assertTrue(((ConnectionQueryServicesImpl) services)
- .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_MIGRATION_TIMESTAMP, mutexRowKey));
+ .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_MIGRATION_TIMESTAMP));
} else {
- ((ConnectionQueryServicesImpl) services).releaseUpgradeMutex(mutexRowKey);
+ services.deleteMutexCell(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE, null, null);
}
}
}
@@ -397,14 +395,13 @@ public class MigrateSystemTablesToSystemNamespaceIT extends BaseTest {
}
}
- // The set will contain SYSMUTEX table since that table is not exposed in SYSCAT
if (systemTablesMapped) {
if (!systemSchemaExists) {
fail(PhoenixDatabaseMetaData.SYSTEM_SCHEMA_NAME + " entry doesn't exist in SYSTEM.CATALOG table.");
}
- assertTrue(namespaceMappedSystemTablesSet.size() == 1);
+ assertTrue(namespaceMappedSystemTablesSet.isEmpty());
} else {
- assertTrue(systemTablesSet.size() == 1);
+ assertTrue(systemTablesSet.isEmpty());
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
index 90f9db6..7d31c62 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryDatabaseMetaDataIT.java
@@ -21,6 +21,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCH
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CHILD_LINK_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
import static org.apache.phoenix.util.TestUtil.ATABLE_NAME;
import static org.apache.phoenix.util.TestUtil.CUSTOM_ENTITY_DATA_FULL_NAME;
@@ -175,6 +176,9 @@ public class QueryDatabaseMetaDataIT extends ParallelStatsDisabledIT {
assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE"));
assertTrue(rs.next());
assertEquals(SYSTEM_CATALOG_SCHEMA, rs.getString("TABLE_SCHEM"));
+ assertEquals(SYSTEM_MUTEX_TABLE_NAME, rs.getString("TABLE_NAME"));
+ assertTrue(rs.next());
+ assertEquals(SYSTEM_CATALOG_SCHEMA, rs.getString("TABLE_SCHEM"));
assertEquals(TYPE_SEQUENCE, rs.getString("TABLE_NAME"));
assertEquals(PTableType.SYSTEM.toString(), rs.getString("TABLE_TYPE"));
assertTrue(rs.next());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableDDLPermissionsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableDDLPermissionsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableDDLPermissionsIT.java
index 8666bb8..86a6b60 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableDDLPermissionsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TableDDLPermissionsIT.java
@@ -20,7 +20,6 @@ import java.security.PrivilegedExceptionAction;
import java.sql.Connection;
import java.util.Collections;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.AuthUtil;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.security.AccessDeniedException;
@@ -54,6 +53,10 @@ public class TableDDLPermissionsIT extends BasePermissionsIT {
Action.READ, Action.EXEC);
grantPermissions(unprivilegedUser.getShortName(), Collections.singleton("SYSTEM:SEQUENCE"), Action.WRITE,
Action.READ, Action.EXEC);
+ grantPermissions(regularUser1.getShortName(), Collections.singleton("SYSTEM:MUTEX"), Action.WRITE,
+ Action.READ, Action.EXEC);
+ grantPermissions(unprivilegedUser.getShortName(), Collections.singleton("SYSTEM:MUTEX"), Action.WRITE,
+ Action.READ, Action.EXEC);
} else {
grantPermissions(regularUser1.getShortName(), PHOENIX_SYSTEM_TABLES, Action.READ, Action.EXEC);
@@ -64,6 +67,10 @@ public class TableDDLPermissionsIT extends BasePermissionsIT {
Action.READ, Action.EXEC);
grantPermissions(unprivilegedUser.getShortName(), Collections.singleton("SYSTEM:SEQUENCE"), Action.WRITE,
Action.READ, Action.EXEC);
+ grantPermissions(regularUser1.getShortName(), Collections.singleton("SYSTEM.MUTEX"), Action.WRITE,
+ Action.READ, Action.EXEC);
+ grantPermissions(unprivilegedUser.getShortName(), Collections.singleton("SYSTEM.MUTEX"), Action.WRITE,
+ Action.READ, Action.EXEC);
}
} catch (Throwable e) {
if (e instanceof Exception) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
index dd6f7f7..956b43c 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/TenantSpecificTablesDDLIT.java
@@ -498,6 +498,8 @@ public class TenantSpecificTablesDDLIT extends BaseTenantSpecificTablesIT {
assertTrue(rs.next());
assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA, PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE, PTableType.SYSTEM);
assertTrue(rs.next());
+ assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA, PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME, PTableType.SYSTEM);
+ assertTrue(rs.next());
assertTableMetaData(rs, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA, PhoenixDatabaseMetaData.TYPE_SEQUENCE, PTableType.SYSTEM);
assertTrue(rs.next());
assertTableMetaData(rs, SYSTEM_CATALOG_SCHEMA, PhoenixDatabaseMetaData.SYSTEM_STATS_TABLE, PTableType.SYSTEM);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
index 2b866a5..c2cf9e6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpgradeIT.java
@@ -18,8 +18,6 @@
package org.apache.phoenix.end2end;
import static com.google.common.base.Preconditions.checkNotNull;
-import static org.apache.phoenix.query.ConnectionQueryServicesImpl.UPGRADE_MUTEX;
-import static org.apache.phoenix.query.ConnectionQueryServicesImpl.UPGRADE_MUTEX_UNLOCKED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
@@ -42,8 +40,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.curator.shaded.com.google.common.collect.Sets;
import org.apache.hadoop.hbase.client.HBaseAdmin;
-import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
@@ -425,39 +421,21 @@ public class UpgradeIT extends ParallelStatsDisabledIT {
}
}
- private void putUnlockKVInSysMutex(byte[] row) throws Exception {
- try (Connection conn = getConnection(false, null)) {
- ConnectionQueryServices services = conn.unwrap(PhoenixConnection.class).getQueryServices();
- try (HTableInterface sysMutexTable = services.getTable(PhoenixDatabaseMetaData.SYSTEM_MUTEX_NAME_BYTES)) {
- byte[] family = PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
- byte[] qualifier = UPGRADE_MUTEX;
- Put put = new Put(row);
- put.addColumn(family, qualifier, UPGRADE_MUTEX_UNLOCKED);
- sysMutexTable.put(put);
- sysMutexTable.flushCommits();
- }
- }
- }
-
@Test
public void testAcquiringAndReleasingUpgradeMutex() throws Exception {
ConnectionQueryServices services = null;
- byte[] mutexRowKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
- generateUniqueName());
try (Connection conn = getConnection(false, null)) {
services = conn.unwrap(PhoenixConnection.class).getQueryServices();
- putUnlockKVInSysMutex(mutexRowKey);
assertTrue(((ConnectionQueryServicesImpl)services)
- .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0, mutexRowKey));
+ .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0));
try {
((ConnectionQueryServicesImpl)services)
- .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0, mutexRowKey);
+ .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
fail();
} catch (UpgradeInProgressException expected) {
}
- assertTrue(((ConnectionQueryServicesImpl)services).releaseUpgradeMutex(mutexRowKey));
- assertFalse(((ConnectionQueryServicesImpl)services).releaseUpgradeMutex(mutexRowKey));
+ ((ConnectionQueryServicesImpl)services).releaseUpgradeMutex();
}
}
@@ -471,7 +449,6 @@ public class UpgradeIT extends ParallelStatsDisabledIT {
final byte[] mutexKey = Bytes.toBytes(generateUniqueName());
try (Connection conn = getConnection(false, null)) {
services = conn.unwrap(PhoenixConnection.class).getQueryServices();
- putUnlockKVInSysMutex(mutexKey);
FutureTask<Void> task1 = new FutureTask<>(new AcquireMutexRunnable(mutexStatus1, services, latch, numExceptions, mutexKey));
FutureTask<Void> task2 = new FutureTask<>(new AcquireMutexRunnable(mutexStatus2, services, latch, numExceptions, mutexKey));
Thread t1 = new Thread(task1);
@@ -509,12 +486,15 @@ public class UpgradeIT extends ParallelStatsDisabledIT {
public Void call() throws Exception {
try {
((ConnectionQueryServicesImpl)services).acquireUpgradeMutex(
- MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0, mutexRowKey);
+ MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0);
acquireStatus.set(true);
} catch (UpgradeInProgressException e) {
numExceptions.incrementAndGet();
} finally {
latch.countDown();
+ if (acquireStatus.get()) {
+ ((ConnectionQueryServicesImpl)services).releaseUpgradeMutex();
+ }
}
return null;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
index fdfd75b..fda9490 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
@@ -20,11 +20,13 @@ package org.apache.phoenix.end2end;
import static com.google.common.collect.Lists.newArrayListWithExpectedSize;
import static org.apache.phoenix.exception.SQLExceptionCode.CANNOT_MODIFY_VIEW_PK;
import static org.apache.phoenix.exception.SQLExceptionCode.NOT_NULLABLE_COLUMN_IN_ROW_KEY;
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.apache.phoenix.util.TestUtil.analyzeTable;
import static org.apache.phoenix.util.TestUtil.getAllSplits;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -43,6 +45,13 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.hadoop.hbase.DoNotRetryIOException;
@@ -52,9 +61,9 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
@@ -67,6 +76,7 @@ import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.ColumnAlreadyExistsException;
+import org.apache.phoenix.schema.ConcurrentTableMutationException;
import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableType;
@@ -74,6 +84,7 @@ import org.apache.phoenix.schema.ReadOnlyTableException;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
@@ -93,6 +104,17 @@ public class ViewIT extends SplitSystemCatalogIT {
protected String tableDDLOptions;
protected boolean transactional;
+
+ private static final String FAILED_VIEWNAME = "FAILED_VIEW";
+ private static final byte[] FAILED_ROWKEY_BYTES =
+ SchemaUtil.getTableKey(null, Bytes.toBytes(SCHEMA2), Bytes.toBytes(FAILED_VIEWNAME));
+ private static final String SLOW_VIEWNAME_PREFIX = "SLOW_VIEW";
+ private static final byte[] SLOW_ROWKEY_PREFIX_BYTES =
+ SchemaUtil.getTableKey(null, Bytes.toBytes(SCHEMA2),
+ Bytes.toBytes(SLOW_VIEWNAME_PREFIX));
+
+ private static volatile CountDownLatch latch1 = null;
+ private static volatile CountDownLatch latch2 = null;
public ViewIT(boolean transactional) {
StringBuilder optionBuilder = new StringBuilder();
@@ -114,7 +136,7 @@ public class ViewIT extends SplitSystemCatalogIT {
Map<String, String> props = Collections.emptyMap();
boolean splitSystemCatalog = (driver == null);
Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(1);
- serverProps.put("hbase.coprocessor.region.classes", FailingRegionObserver.class.getName());
+ serverProps.put("hbase.coprocessor.region.classes", TestMetaDataRegionObserver.class.getName());
serverProps.put("hbase.coprocessor.abortonerror", "false");
setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(props.entrySet().iterator()));
// Split SYSTEM.CATALOG once after the mini-cluster is started
@@ -123,6 +145,49 @@ public class ViewIT extends SplitSystemCatalogIT {
}
}
+ public static class TestMetaDataRegionObserver extends BaseRegionObserver {
+ @Override
+ public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
+ if (shouldFail(c, miniBatchOp.getOperation(0))) {
+ // throwing anything other than instances of IOException result
+ // in this coprocessor being unloaded
+ // DoNotRetryIOException tells HBase not to retry this mutation
+ // multiple times
+ throw new DoNotRetryIOException();
+ } else if (shouldSlowDown(c, miniBatchOp.getOperation(0))) {
+ // simulate a slow write to SYSTEM.CATALOG
+ if (latch1 != null) {
+ latch1.countDown();
+ }
+ if (latch2 != null) {
+ try {
+ // wait till the second task is complete before completing the first task
+ boolean result = latch2.await(2, TimeUnit.MINUTES);
+ if (!result) {
+ throw new RuntimeException("Second task took took long to complete");
+ }
+ } catch (InterruptedException e) {
+ }
+ }
+ }
+ }
+
+ private boolean shouldFail(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m) {
+ TableName tableName = c.getEnvironment().getRegion().getRegionInfo().getTable();
+ return tableName.equals(PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
+ && (Bytes.equals(FAILED_ROWKEY_BYTES, m.getRow()));
+ }
+
+ private boolean shouldSlowDown(ObserverContext<RegionCoprocessorEnvironment> c,
+ Mutation m) {
+ TableName tableName = c.getEnvironment().getRegion().getRegionInfo().getTable();
+ byte[] rowKeyPrefix = Arrays.copyOf(m.getRow(), SLOW_ROWKEY_PREFIX_BYTES.length);
+ return tableName.equals(PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
+ && (Bytes.equals(SLOW_ROWKEY_PREFIX_BYTES, rowKeyPrefix));
+ }
+ }
+
@Test
public void testReadOnlyOnUpdatableView() throws Exception {
String fullTableName = SchemaUtil.getTableName(SCHEMA1, generateUniqueName());
@@ -1274,28 +1339,160 @@ public class ViewIT extends SplitSystemCatalogIT {
PhoenixRuntime.getTableNoCache(conn, fullViewName2);
}
- private static final String FAILED_VIEWNAME = "FAILED_VIEW";
- private static final byte[] ROWKEY_TO_FAIL_BYTES = SchemaUtil.getTableKey(null, Bytes.toBytes(SCHEMA2),
- Bytes.toBytes(FAILED_VIEWNAME));
-
- public static class FailingRegionObserver extends SimpleRegionObserver {
+ @Test
+ public void testConcurrentViewCreationAndTableDrop() throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ String fullTableName = SchemaUtil.getTableName(SCHEMA1, generateUniqueName());
+ String fullViewName1 =
+ SchemaUtil.getTableName(SCHEMA2,
+ SLOW_VIEWNAME_PREFIX + "_" + generateUniqueName());
+ String fullViewName2 = SchemaUtil.getTableName(SCHEMA3, generateUniqueName());
+ latch1 = new CountDownLatch(1);
+ latch2 = new CountDownLatch(1);
+ String tableDdl =
+ "CREATE TABLE " + fullTableName + " (k INTEGER NOT NULL PRIMARY KEY, v1 DATE)"
+ + tableDDLOptions;
+ conn.createStatement().execute(tableDdl);
+
+ ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ t.setPriority(Thread.MIN_PRIORITY);
+ return t;
+ }
+ });
+
+ // create the view in a separate thread (which will take some time
+ // to complete)
+ Future<Exception> future =
+ executorService.submit(new CreateViewRunnable(fullTableName, fullViewName1));
+ // wait till the thread makes the rpc to create the view
+ latch1.await();
+ tableDdl = "DROP TABLE " + fullTableName;
+ try {
+ // drop table should fail as we are concurrently adding a view
+ conn.createStatement().execute(tableDdl);
+ fail("Creating a view while concurrently dropping the base table should fail");
+ } catch (ConcurrentTableMutationException e) {
+ }
+ latch2.countDown();
+
+ Exception e = future.get();
+ assertNull(e);
+
+ // create another view to ensure that the cell used to prevent
+ // concurrent modifications was removed
+ String ddl =
+ "CREATE VIEW " + fullViewName2 + " (v2 VARCHAR) AS SELECT * FROM "
+ + fullTableName + " WHERE k = 6";
+ conn.createStatement().execute(ddl);
+ }
+ }
+
+ @Test
+ public void testConcurrentAddColumn() throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ String fullTableName = SchemaUtil.getTableName(SCHEMA1, generateUniqueName());
+ String fullViewName1 =
+ SchemaUtil.getTableName(SCHEMA2,
+ SLOW_VIEWNAME_PREFIX + "_" + generateUniqueName());
+ String fullViewName2 = SchemaUtil.getTableName(SCHEMA3, generateUniqueName());
+ // create base table
+ String tableDdl =
+ "CREATE TABLE " + fullTableName + " (k INTEGER NOT NULL PRIMARY KEY, v1 DATE)"
+ + tableDDLOptions;
+ conn.createStatement().execute(tableDdl);
+ // create a view
+ String ddl =
+ "CREATE VIEW " + fullViewName1 + " (v2 VARCHAR) AS SELECT * FROM "
+ + fullTableName + " WHERE k = 6";
+ conn.createStatement().execute(ddl);
+
+ latch1 = new CountDownLatch(1);
+ latch2 = new CountDownLatch(1);
+ ExecutorService executorService = Executors.newFixedThreadPool(1, new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setDaemon(true);
+ t.setPriority(Thread.MIN_PRIORITY);
+ return t;
+ }
+ });
+
+ // add a column to the view in a separate thread (which will take
+ // some time to complete)
+ Future<Exception> future = executorService.submit(new AddColumnRunnable(fullViewName1));
+ // wait till the thread makes the rpc to create the view
+ boolean result = latch1.await(2, TimeUnit.MINUTES);
+ if (!result) {
+ fail("The create view rpc look too long");
+ }
+ tableDdl = "ALTER TABLE " + fullTableName + " ADD v3 INTEGER";
+ try {
+ // add the same column to the base table with a different type
+ conn.createStatement().execute(tableDdl);
+ fail("Creating a view while concurrently dropping the base table should fail");
+ } catch (ConcurrentTableMutationException e) {
+ }
+ latch2.countDown();
+
+ Exception e = future.get();
+ assertNull(e);
+
+ // add a the same column to the another view to ensure that the cell used
+ // to prevent concurrent modifications was removed
+ ddl = "CREATE VIEW " + fullViewName2 + " (v2 VARCHAR) AS SELECT * FROM "
+ + fullTableName + " WHERE k = 6";
+ conn.createStatement().execute(ddl);
+ tableDdl = "ALTER VIEW " + fullViewName2 + " ADD v3 INTEGER";
+ conn.createStatement().execute(tableDdl);
+ }
+ }
+
+ private class CreateViewRunnable implements Callable<Exception> {
+ private final String fullTableName;
+ private final String fullViewName;
+
+ public CreateViewRunnable(String fullTableName, String fullViewName) {
+ this.fullTableName = fullTableName;
+ this.fullViewName = fullViewName;
+ }
+
@Override
- public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
- MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
- if (shouldFail(c, miniBatchOp.getOperation(0))) {
- // throwing anything other than instances of IOException result
- // in this coprocessor being unloaded
- // DoNotRetryIOException tells HBase not to retry this mutation
- // multiple times
- throw new DoNotRetryIOException();
+ public Exception call() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String ddl =
+ "CREATE VIEW " + fullViewName + " (v2 VARCHAR) AS SELECT * FROM "
+ + fullTableName + " WHERE k = 5";
+ conn.createStatement().execute(ddl);
+ } catch (SQLException e) {
+ return e;
}
+ return null;
}
+ }
- private boolean shouldFail(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m) {
- TableName tableName = c.getEnvironment().getRegion().getRegionInfo().getTable();
- return tableName.equals(PhoenixDatabaseMetaData.SYSTEM_CATALOG_HBASE_TABLE_NAME)
- && (Bytes.equals(ROWKEY_TO_FAIL_BYTES, m.getRow()));
+ private class AddColumnRunnable implements Callable<Exception> {
+ private final String fullViewName;
+
+ public AddColumnRunnable(String fullViewName) {
+ this.fullViewName = fullViewName;
}
+ @Override
+ public Exception call() throws Exception {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String ddl = "ALTER VIEW " + fullViewName + " ADD v3 CHAR(15)";
+ conn.createStatement().execute(ddl);
+ } catch (SQLException e) {
+ return e;
+ }
+ return null;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index dab1048..7a4a481 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -331,6 +331,8 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String SEQUENCE_TABLE_TYPE = SYSTEM_SEQUENCE_TABLE;
public static final String SYNC_INDEX_CREATED_DATE = "SYNC_INDEX_CREATED_DATE";
+ public static final String SYSTEM_MUTEX_COLUMN_NAME = "MUTEX_VALUE";
+ public static final byte[] SYSTEM_MUTEX_COLUMN_NAME_BYTES = Bytes.toBytes(SYSTEM_MUTEX_COLUMN_NAME);
public static final String SYSTEM_MUTEX_TABLE_NAME = "MUTEX";
public static final String SYSTEM_MUTEX_NAME = SchemaUtil.getTableName(QueryConstants.SYSTEM_SCHEMA_NAME, SYSTEM_MUTEX_TABLE_NAME);
public static final TableName SYSTEM_MUTEX_HBASE_TABLE_NAME = TableName.valueOf(SYSTEM_MUTEX_NAME);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
index 6f8cbc0..0820232 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
@@ -169,4 +169,18 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated
public QueryLoggerDisruptor getQueryDisruptor();
public PhoenixTransactionClient initTransactionClient(TransactionFactory.Provider provider);
+
+ /**
+ * Writes a cell to SYSTEM.MUTEX using checkAndPut to ensure only a single client can execute a
+ * particular task. The params are used to generate the rowkey.
+ * @return true if this client was able to successfully acquire the mutex
+ */
+ public boolean writeMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException;
+
+ /**
+ * Deletes a cell that was written to SYSTEM.MUTEX. The params are used to generate the rowkey.
+ */
+ public void deleteMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 4c7630d..8bbb379 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -64,6 +64,7 @@ import static org.apache.phoenix.util.UpgradeUtil.moveChildLinks;
import static org.apache.phoenix.util.UpgradeUtil.upgradeTo4_5_0;
import java.io.IOException;
+import java.lang.management.ManagementFactory;
import java.lang.ref.WeakReference;
import java.sql.PreparedStatement;
import java.sql.ResultSetMetaData;
@@ -109,6 +110,7 @@ import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Append;
+import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -326,9 +328,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
private final AtomicBoolean upgradeRequired = new AtomicBoolean(false);
private final int maxConnectionsAllowed;
private final boolean shouldThrottleNumConnections;
- public static final byte[] UPGRADE_MUTEX = "UPGRADE_MUTEX".getBytes();
- public static final byte[] UPGRADE_MUTEX_LOCKED = "UPGRADE_MUTEX_LOCKED".getBytes();
- public static final byte[] UPGRADE_MUTEX_UNLOCKED = "UPGRADE_MUTEX_UNLOCKED".getBytes();
+ public static final byte[] MUTEX_LOCKED = "MUTEX_LOCKED".getBytes();
private static interface FeatureSupported {
boolean isSupported(ConnectionQueryServices services);
@@ -2502,6 +2502,10 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
protected String getChildLinkDDL() {
return setSystemDDLProperties(QueryConstants.CREATE_CHILD_LINK_METADATA);
}
+
+ protected String getMutexDDL() {
+ return setSystemDDLProperties(QueryConstants.CREATE_MUTEX_METADTA);
+ }
private String setSystemDDLProperties(String ddl) {
return String.format(ddl,
@@ -2673,13 +2677,6 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
columnDesc.setTimeToLive(TTL_FOR_MUTEX); // Let mutex expire after some time
tableDesc.addFamily(columnDesc);
admin.createTable(tableDesc);
- try (HTableInterface sysMutexTable = getTable(mutexTableName.getName())) {
- byte[] mutexRowKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE);
- Put put = new Put(mutexRowKey);
- put.addColumn(PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES, UPGRADE_MUTEX, UPGRADE_MUTEX_UNLOCKED);
- sysMutexTable.put(put);
- }
} catch (IOException e) {
if(!Iterables.isEmpty(Iterables.filter(Throwables.getCausalChain(e), AccessDeniedException.class)) ||
!Iterables.isEmpty(Iterables.filter(Throwables.getCausalChain(e), org.apache.hadoop.hbase.TableNotFoundException.class))) {
@@ -2687,13 +2684,6 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
} else {
throw e;
}
- }catch(PhoenixIOException e){
- if(e.getCause()!=null && e.getCause() instanceof AccessDeniedException)
- {
- //Ignore
- }else{
- throw e;
- }
}
}
@@ -2723,13 +2713,10 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
try {
metaConnection.createStatement().executeUpdate(getChildLinkDDL());
} catch (TableAlreadyExistsException e) {}
- // Catch the IOException to log the error message and then bubble it up for the client to retry.
try {
- createSysMutexTableIfNotExists(hbaseAdmin);
- } catch (IOException exception) {
- logger.error("Failed to created SYSMUTEX table. Upgrade or migration is not possible without it. Please retry.");
- throw exception;
- }
+ metaConnection.createStatement().executeUpdate(getMutexDDL());
+ } catch (TableAlreadyExistsException e) {}
+ // Catch the IOException to log the error message and then bubble it up for the client to retry.
}
/**
@@ -3022,8 +3009,6 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
String sysCatalogTableName = null;
SQLException toThrow = null;
boolean acquiredMutexLock = false;
- byte[] mutexRowKey = SchemaUtil.getTableKey(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
- PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE);
boolean snapshotCreated = false;
try {
if (!isUpgradeRequired()) {
@@ -3054,7 +3039,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
sysCatalogTableName = SchemaUtil.getPhysicalName(SYSTEM_CATALOG_NAME_BYTES, this.getProps()).getNameAsString();
if (SchemaUtil.isNamespaceMappingEnabled(PTableType.SYSTEM, ConnectionQueryServicesImpl.this.getProps())) {
// Try acquiring a lock in SYSMUTEX table before migrating the tables since it involves disabling the table.
- if (acquiredMutexLock = acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_MIGRATION_TIMESTAMP, mutexRowKey)) {
+ if (acquiredMutexLock = acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_MIGRATION_TIMESTAMP)) {
logger.debug("Acquired lock in SYSMUTEX table for migrating SYSTEM tables to SYSTEM namespace "
+ "and/or upgrading " + sysCatalogTableName);
}
@@ -3073,7 +3058,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
// Try acquiring a lock in SYSMUTEX table before upgrading SYSCAT. If we cannot acquire the lock,
// it means some old client is either migrating SYSTEM tables or trying to upgrade the schema of
// SYSCAT table and hence it should not be interrupted
- if (acquiredMutexLock = acquireUpgradeMutex(currentServerSideTableTimeStamp, mutexRowKey)) {
+ if (acquiredMutexLock = acquireUpgradeMutex(currentServerSideTableTimeStamp)) {
logger.debug("Acquired lock in SYSMUTEX table for upgrading " + sysCatalogTableName);
snapshotName = getSysCatalogSnapshotName(currentServerSideTableTimeStamp);
createSnapshot(snapshotName, sysCatalogTableName);
@@ -3173,6 +3158,9 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
try {
metaConnection.createStatement().executeUpdate(getChildLinkDDL());
} catch (NewerTableAlreadyExistsException e) {} catch (TableAlreadyExistsException e) {}
+ try {
+ metaConnection.createStatement().executeUpdate(getMutexDDL());
+ } catch (NewerTableAlreadyExistsException e) {} catch (TableAlreadyExistsException e) {}
// In case namespace mapping is enabled and system table to system namespace mapping is also enabled,
// create an entry for the SYSTEM namespace in the SYSCAT table, so that GRANT/REVOKE commands can work
@@ -3216,7 +3204,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
} finally {
if (acquiredMutexLock) {
try {
- releaseUpgradeMutex(mutexRowKey);
+ releaseUpgradeMutex();
} catch (IOException e) {
logger.warn("Release of upgrade mutex failed ", e);
}
@@ -3410,17 +3398,10 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
// No tables exist matching "SYSTEM\..*", they are all already in "SYSTEM:.*"
if (tableNames.size() == 0) { return; }
// Try to move any remaining tables matching "SYSTEM\..*" into "SYSTEM:"
- if (tableNames.size() > 5) {
- logger.warn("Expected 5 system tables but found " + tableNames.size() + ":" + tableNames);
+ if (tableNames.size() > 7) {
+ logger.warn("Expected 7 system tables but found " + tableNames.size() + ":" + tableNames);
}
- // Handle the upgrade of SYSMUTEX table separately since it doesn't have any entries in SYSCAT
- logger.info("Migrating SYSTEM.MUTEX table to SYSTEM namespace.");
- String sysMutexSrcTableName = PhoenixDatabaseMetaData.SYSTEM_MUTEX_NAME;
- String sysMutexDestTableName = SchemaUtil.getPhysicalName(sysMutexSrcTableName.getBytes(), this.getProps()).getNameAsString();
- UpgradeUtil.mapTableToNamespace(admin, sysMutexSrcTableName, sysMutexDestTableName, PTableType.SYSTEM);
- tableNames.remove(PhoenixDatabaseMetaData.SYSTEM_MUTEX_HBASE_TABLE_NAME);
-
byte[] mappedSystemTable = SchemaUtil
.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, this.getProps()).getName();
metatable = getTable(mappedSystemTable);
@@ -3464,64 +3445,95 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
* @throws SQLException
*/
@VisibleForTesting
- public boolean acquireUpgradeMutex(long currentServerSideTableTimestamp, byte[] rowToLock) throws IOException,
+ public boolean acquireUpgradeMutex(long currentServerSideTableTimestamp)
+ throws IOException,
SQLException {
Preconditions.checkArgument(currentServerSideTableTimestamp < MIN_SYSTEM_TABLE_TIMESTAMP);
-
byte[] sysMutexPhysicalTableNameBytes = getSysMutexPhysicalTableNameBytes();
if(sysMutexPhysicalTableNameBytes == null) {
throw new UpgradeInProgressException(getVersion(currentServerSideTableTimestamp),
getVersion(MIN_SYSTEM_TABLE_TIMESTAMP));
}
+ if (!writeMutexCell(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE, null, null)) {
+ throw new UpgradeInProgressException(getVersion(currentServerSideTableTimestamp),
+ getVersion(MIN_SYSTEM_TABLE_TIMESTAMP));
+ }
+ return true;
+ }
- try (HTableInterface sysMutexTable = getTable(sysMutexPhysicalTableNameBytes)) {
- byte[] family = PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
- byte[] qualifier = UPGRADE_MUTEX;
- byte[] oldValue = UPGRADE_MUTEX_UNLOCKED;
- byte[] newValue = UPGRADE_MUTEX_LOCKED;
- Put put = new Put(rowToLock);
- put.addColumn(family, qualifier, newValue);
- boolean acquired = sysMutexTable.checkAndPut(rowToLock, family, qualifier, oldValue, put);
- if (!acquired) {
- /*
- * Because of TTL on the SYSTEM_MUTEX_FAMILY, it is very much possible that the cell
- * has gone away. So we need to retry with an old value of null. Note there is a small
- * race condition here that between the two checkAndPut calls, it is possible that another
- * request would have set the value back to UPGRADE_MUTEX_UNLOCKED. In that scenario this
- * following checkAndPut would still return false even though the lock was available.
- */
- acquired = sysMutexTable.checkAndPut(rowToLock, family, qualifier, null, put);
- if (!acquired) {
- throw new UpgradeInProgressException(getVersion(currentServerSideTableTimestamp),
- getVersion(MIN_SYSTEM_TABLE_TIMESTAMP));
+ @Override
+ public boolean writeMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ try {
+ byte[] rowKey =
+ columnName != null
+ ? SchemaUtil.getColumnKey(tenantId, schemaName, tableName, columnName,
+ familyName)
+ : SchemaUtil.getTableKey(tenantId, schemaName, tableName);
+ // at this point the system mutex table should have been created or
+ // an exception thrown
+ byte[] sysMutexPhysicalTableNameBytes = getSysMutexPhysicalTableNameBytes();
+ try (HTableInterface sysMutexTable = getTable(sysMutexPhysicalTableNameBytes)) {
+ byte[] family = PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
+ byte[] qualifier = PhoenixDatabaseMetaData.SYSTEM_MUTEX_COLUMN_NAME_BYTES;
+ byte[] value = MUTEX_LOCKED;
+ Put put = new Put(rowKey);
+ put.addColumn(family, qualifier, value);
+ boolean checkAndPut =
+ sysMutexTable.checkAndPut(rowKey, family, qualifier, null, put);
+ String processName = ManagementFactory.getRuntimeMXBean().getName();
+ String msg =
+ " tenantId : " + tenantId + " schemaName : " + schemaName + " tableName : "
+ + tableName + " columnName : " + columnName + " familyName : "
+ + familyName;
+ if (!checkAndPut) {
+ logger.error(processName + " failed to acquire mutex for "+ msg);
+ }
+ else {
+ logger.debug(processName + " acquired mutex for "+ msg);
}
+ return checkAndPut;
}
- return true;
+ } catch (IOException e) {
+ throw ServerUtil.parseServerException(e);
}
}
@VisibleForTesting
- public boolean releaseUpgradeMutex(byte[] mutexRowKey) throws IOException, SQLException {
- boolean released = false;
-
- byte[] sysMutexPhysicalTableNameBytes = getSysMutexPhysicalTableNameBytes();
- if(sysMutexPhysicalTableNameBytes == null) {
- // We shouldn't never be really in this situation where neither SYSMUTEX or SYS:MUTEX exists
- return true;
- }
+ public void releaseUpgradeMutex() throws IOException, SQLException {
+ deleteMutexCell(null, PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA,
+ PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE, null, null);
+ }
- try (HTableInterface sysMutexTable = getTable(sysMutexPhysicalTableNameBytes)) {
- byte[] family = PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
- byte[] qualifier = UPGRADE_MUTEX;
- byte[] expectedValue = UPGRADE_MUTEX_LOCKED;
- byte[] newValue = UPGRADE_MUTEX_UNLOCKED;
- Put put = new Put(mutexRowKey);
- put.addColumn(family, qualifier, newValue);
- released = sysMutexTable.checkAndPut(mutexRowKey, family, qualifier, expectedValue, put);
- } catch (Exception e) {
- logger.warn("Release of upgrade mutex failed", e);
+ @Override
+ public void deleteMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ try {
+ byte[] rowKey =
+ columnName != null
+ ? SchemaUtil.getColumnKey(tenantId, schemaName, tableName, columnName,
+ familyName)
+ : SchemaUtil.getTableKey(tenantId, schemaName, tableName);
+ // at this point the system mutex table should have been created or
+ // an exception thrown
+ byte[] sysMutexPhysicalTableNameBytes = getSysMutexPhysicalTableNameBytes();
+ try (HTableInterface sysMutexTable = getTable(sysMutexPhysicalTableNameBytes)) {
+ byte[] family = PhoenixDatabaseMetaData.SYSTEM_MUTEX_FAMILY_NAME_BYTES;
+ byte[] qualifier = PhoenixDatabaseMetaData.SYSTEM_MUTEX_COLUMN_NAME_BYTES;
+ Delete delete = new Delete(rowKey);
+ delete.addColumn(family, qualifier);
+ sysMutexTable.delete(delete);
+ String processName = ManagementFactory.getRuntimeMXBean().getName();
+ String msg =
+ " tenantId : " + tenantId + " schemaName : " + schemaName + " tableName : "
+ + tableName + " columnName : " + columnName + " familyName : "
+ + familyName;
+ logger.debug(processName + " released mutex for "+ msg);
+ }
+ } catch (IOException e) {
+ throw ServerUtil.parseServerException(e);
}
- return released;
}
private byte[] getSysMutexPhysicalTableNameBytes() throws IOException, SQLException {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index 5a46214..f088172 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -180,6 +180,10 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
protected String getChildLinkDDL() {
return setSystemDDLProperties(QueryConstants.CREATE_CHILD_LINK_METADATA);
}
+
+ protected String getMutexDDL() {
+ return setSystemDDLProperties(QueryConstants.CREATE_MUTEX_METADTA);
+ }
private String setSystemDDLProperties(String ddl) {
return String.format(ddl,
@@ -379,6 +383,11 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
.executeUpdate(getChildLinkDDL());
} catch (NewerTableAlreadyExistsException ignore) {
}
+ try {
+ metaConnection.createStatement()
+ .executeUpdate(getMutexDDL());
+ } catch (NewerTableAlreadyExistsException ignore) {
+ }
} catch (SQLException e) {
sqlE = e;
} finally {
@@ -730,4 +739,15 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
public PhoenixTransactionClient initTransactionClient(Provider provider) {
return null; // Client is not necessary
}
+
+ @Override
+ public boolean writeMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ return true;
+ }
+
+ @Override
+ public void deleteMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
index b3e2cb2..147e873 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
@@ -370,4 +370,15 @@ public class DelegateConnectionQueryServices extends DelegateQueryServices imple
public PhoenixTransactionClient initTransactionClient(Provider provider) {
return getDelegate().initTransactionClient(provider);
}
+
+ @Override
+ public boolean writeMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ return true;
+ }
+
+ @Override
+ public void deleteMutexCell(String tenantId, String schemaName, String tableName,
+ String columnName, String familyName) throws SQLException {
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 8d8d47f..32fedc8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -355,5 +355,19 @@ public interface QueryConstants {
+ "," + COLUMN_NAME + "," + COLUMN_FAMILY + "))\n" + HConstants.VERSIONS + "=%s,\n"
+ HColumnDescriptor.KEEP_DELETED_CELLS + "=%s,\n" + PhoenixDatabaseMetaData.TRANSACTIONAL + "="
+ Boolean.FALSE;
+
+ public static final String CREATE_MUTEX_METADTA =
+ "CREATE IMMUTABLE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_MUTEX_TABLE_NAME + "\"(\n" +
+ // Pk columns
+ TENANT_ID + " VARCHAR NULL," +
+ TABLE_SCHEM + " VARCHAR NULL," +
+ TABLE_NAME + " VARCHAR NOT NULL," +
+ COLUMN_NAME + " VARCHAR NULL," + // null for table row
+ COLUMN_FAMILY + " VARCHAR NULL " + // using for CF to uniqueness for columns
+ "CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + ","
+ + TABLE_SCHEM + "," + TABLE_NAME + "," + COLUMN_NAME + "," + COLUMN_FAMILY + "))\n" +
+ HConstants.VERSIONS + "=%s,\n" +
+ HColumnDescriptor.KEEP_DELETED_CELLS + "=%s,\n" +
+ PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c9b8d114/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 625d03f..c714eab 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -1874,6 +1874,24 @@ public class MetaDataClient {
}
return false;
}
+
+ /**
+ * If we are creating a view we write a cell to the SYSTEM.MUTEX table with the rowkey of the
+ * parent table to prevent concurrent modifications
+ */
+ private boolean writeCell(String tenantId, String schemaName, String tableName, String columnName)
+ throws SQLException {
+ return connection.getQueryServices().writeMutexCell(tenantId, schemaName, tableName, columnName, null);
+ }
+
+ /**
+ * Remove the cell that was written to to the SYSTEM.MUTEX table with the rowkey of the
+ * parent table to prevent concurrent modifications
+ */
+ private void deleteCell(String tenantId, String schemaName, String tableName, String columnName)
+ throws SQLException {
+ connection.getQueryServices().deleteMutexCell(tenantId, schemaName, tableName, columnName, null);
+ }
private PTable createTableInternal(CreateTableStatement statement, byte[][] splits,
final PTable parent, String viewStatement, ViewType viewType,
@@ -1884,6 +1902,7 @@ public class MetaDataClient {
final PTableType tableType = statement.getTableType();
boolean wasAutoCommit = connection.getAutoCommit();
connection.rollback();
+ boolean acquiredMutex = false;
try {
connection.setAutoCommit(false);
List<Mutation> tableMetaData = Lists.newArrayListWithExpectedSize(statement.getColumnDefs().size() + 3);
@@ -1913,6 +1932,21 @@ public class MetaDataClient {
boolean isLocalIndex = indexType == IndexType.LOCAL;
QualifierEncodingScheme encodingScheme = NON_ENCODED_QUALIFIERS;
ImmutableStorageScheme immutableStorageScheme = ONE_CELL_PER_COLUMN;
+
+ if (tableType == PTableType.VIEW) {
+ PName physicalName = parent.getPhysicalName();
+ String physicalSchemaName =
+ SchemaUtil.getSchemaNameFromFullName(physicalName.getString());
+ String physicalTableName =
+ SchemaUtil.getTableNameFromFullName(physicalName.getString());
+ // acquire the mutex using the global physical table name to
+ // prevent creating views while concurrently dropping the base
+ // table
+ acquiredMutex = writeCell(null, physicalSchemaName, physicalTableName, null);
+ if (!acquiredMutex) {
+ throw new ConcurrentTableMutationException(physicalSchemaName, physicalTableName);
+ }
+ }
if (parent != null && tableType == PTableType.INDEX) {
timestamp = TransactionUtil.getTableTimestamp(connection, transactionProvider != null, transactionProvider);
storeNulls = parent.getStoreNulls();
@@ -2833,6 +2867,16 @@ public class MetaDataClient {
}
} finally {
connection.setAutoCommit(wasAutoCommit);
+ if (acquiredMutex && tableType == PTableType.VIEW) {
+ PName physicalName = parent.getPhysicalName();
+ String physicalSchemaName =
+ SchemaUtil.getSchemaNameFromFullName(physicalName.getString());
+ String physicalTableName =
+ SchemaUtil.getTableNameFromFullName(physicalName.getString());
+ // releasing mutex on the table (required to prevent creating views while concurrently
+ // dropping the base table)
+ deleteCell(null, physicalSchemaName, physicalTableName, null);
+ }
}
}
@@ -2942,9 +2986,11 @@ public class MetaDataClient {
boolean ifExists, boolean cascade, boolean skipAddingParentColumns) throws SQLException {
connection.rollback();
boolean wasAutoCommit = connection.getAutoCommit();
+ PName tenantId = connection.getTenantId();
+ String tenantIdStr = tenantId == null ? null : tenantId.getString();
+ boolean acquiredMutex = false;
+ String physicalTableName = SchemaUtil.getTableName(schemaName, tableName);
try {
- PName tenantId = connection.getTenantId();
- String tenantIdStr = tenantId == null ? null : tenantId.getString();
byte[] key = SchemaUtil.getTableKey(tenantIdStr, schemaName, tableName);
Long scn = connection.getSCN();
long clientTimeStamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
@@ -2957,6 +3003,14 @@ public class MetaDataClient {
Delete linkDelete = new Delete(linkKey, clientTimeStamp);
tableMetaData.add(linkDelete);
}
+ if (tableType == PTableType.TABLE) {
+ // acquire a mutex on the table to prevent creating views while concurrently
+ // dropping the base table
+ acquiredMutex = writeCell(null, schemaName, tableName, null);
+ if (!acquiredMutex) {
+ throw new ConcurrentTableMutationException(schemaName, schemaName);
+ }
+ }
MetaDataMutationResult result = connection.getQueryServices().dropTable(tableMetaData, tableType, cascade, skipAddingParentColumns);
MutationCode code = result.getMutationCode();
PTable table = result.getTable();
@@ -3034,6 +3088,11 @@ public class MetaDataClient {
return new MutationState(0, 0, connection);
} finally {
connection.setAutoCommit(wasAutoCommit);
+ // releasing mutex on the table (required to prevent creating views while concurrently
+ // dropping the base table)
+ if (acquiredMutex && tableType == PTableType.TABLE) {
+ deleteCell(null, schemaName, tableName, null);
+ }
}
}
@@ -3252,11 +3311,18 @@ public class MetaDataClient {
throws SQLException {
connection.rollback();
boolean wasAutoCommit = connection.getAutoCommit();
+ List<PColumn> columns = Lists.newArrayListWithExpectedSize(origColumnDefs != null ? origColumnDefs.size() : 0);
+ PName tenantId = connection.getTenantId();
+ String schemaName = table.getSchemaName().getString();
+ String tableName = table.getTableName().getString();
+ PName physicalName = table.getPhysicalName();
+ String physicalSchemaName =
+ SchemaUtil.getSchemaNameFromFullName(physicalName.getString());
+ String physicalTableName =
+ SchemaUtil.getTableNameFromFullName(physicalName.getString());
+ Set<String> acquiredColumnMutexSet = Sets.newHashSetWithExpectedSize(3);
try {
connection.setAutoCommit(false);
- PName tenantId = connection.getTenantId();
- String schemaName = table.getSchemaName().getString();
- String tableName = table.getTableName().getString();
List<ColumnDef> columnDefs = null;
if (table.isAppendOnlySchema()) {
@@ -3337,7 +3403,6 @@ public class MetaDataClient {
boolean willBeTxnl = metaProperties.getNonTxToTx();
Long timeStamp = TransactionUtil.getTableTimestamp(connection, table.isTransactional() || willBeTxnl, table.isTransactional() ? table.getTransactionProvider() : metaPropertiesEvaluated.getTransactionProvider());
int numPkColumnsAdded = 0;
- List<PColumn> columns = Lists.newArrayListWithExpectedSize(numCols);
Set<String> colFamiliesForPColumnsToBeAdded = new LinkedHashSet<>();
Set<String> families = new LinkedHashSet<>();
PTable tableForCQCounters = tableType == PTableType.VIEW ? PhoenixRuntime.getTable(connection, table.getPhysicalName().getString()) : table;
@@ -3534,6 +3599,18 @@ public class MetaDataClient {
}
}
+ boolean acquiredMutex = true;
+ for (PColumn pColumn : columns) {
+ // acquire the mutex using the global physical table name to
+ // prevent creating the same column on a table or view with
+ // a conflicting type etc
+ acquiredMutex = writeCell(null, physicalSchemaName, physicalTableName,
+ pColumn.getName().getString());
+ if (!acquiredMutex) {
+ throw new ConcurrentTableMutationException(physicalSchemaName, physicalTableName);
+ }
+ acquiredColumnMutexSet.add(pColumn.getName().getString());
+ }
MetaDataMutationResult result = connection.getQueryServices().addColumn(tableMetaData, table, properties, colFamiliesForPColumnsToBeAdded, columns);
try {
MutationCode code = processMutationResult(schemaName, tableName, result);
@@ -3604,6 +3681,12 @@ public class MetaDataClient {
}
} finally {
connection.setAutoCommit(wasAutoCommit);
+ if (!acquiredColumnMutexSet.isEmpty()) {
+ for (String columnName : acquiredColumnMutexSet) {
+ // release the mutex (used to prevent concurrent conflicting add column changes)
+ deleteCell(null, physicalSchemaName, physicalTableName, columnName);
+ }
+ }
}
}