phoenix-core/src/it/java/org/apache/phoenix/end2end/AppendOnlySchemaIT.java - phoenix - Git at Google

 /*
  * 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.phoenix.end2end;

 import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyList;
 import static org.mockito.Matchers.anyListOf;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyMap;
 import static org.mockito.Matchers.anySetOf;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Matchers.isNull;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;

 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;

 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.exception.SQLExceptionCode;
 import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.jdbc.PhoenixEmbeddedDriver;
 import org.apache.phoenix.query.ConnectionQueryServices;
 import org.apache.phoenix.schema.ColumnAlreadyExistsException;
 import org.apache.phoenix.schema.PColumn;
 import org.apache.phoenix.schema.PName;
 import org.apache.phoenix.schema.PTable;
 import org.apache.phoenix.schema.PTableKey;
 import org.apache.phoenix.schema.PTableType;
 import org.apache.phoenix.schema.TableAlreadyExistsException;
 import org.apache.phoenix.util.PropertiesUtil;
 import org.junit.Test;
 import org.mockito.Mockito;

 public class AppendOnlySchemaIT extends ParallelStatsDisabledIT {


     private void testTableWithSameSchema(boolean notExists, boolean sameClient) throws Exception {

         // use a spyed ConnectionQueryServices so we can verify calls to getTable
         ConnectionQueryServices connectionQueryServices =
                 Mockito.spy(driver.getConnectionQueryServices(getUrl(),
                     PropertiesUtil.deepCopy(TEST_PROPERTIES)));
         Properties props = new Properties();
         props.putAll(PhoenixEmbeddedDriver.DEFFAULT_PROPS.asMap());

         try (Connection conn1 = connectionQueryServices.connect(getUrl(), props);
                 Connection conn2 = sameClient ? conn1 : connectionQueryServices.connect(getUrl(), props)) {

             String metricTableName = generateUniqueName();
             String viewName = generateUniqueName();
             String metricIdSeqTableName = generateUniqueName();
             // create sequence for auto partition
             conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + " CACHE 1");
             // create base table
             conn1.createStatement().execute("CREATE TABLE "+ metricTableName + "(metricId INTEGER NOT NULL, metricVal DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
                     + " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
             // create view
             String ddl =
                     "CREATE VIEW " + (notExists ? "IF NOT EXISTS " : "")
                             + viewName + " ( hostName varchar NOT NULL, tagName varChar"
                             + " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
                             + " AS SELECT * FROM " + metricTableName
                             + " UPDATE_CACHE_FREQUENCY=300000";
             conn1.createStatement().execute(ddl);
             conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host1', 1.0)");
             conn1.commit();
             reset(connectionQueryServices);

             // execute same create ddl
             try {
                 conn2.createStatement().execute(ddl);
                 if (!notExists) {
                     fail("Create Table should fail");
                 }
             }
             catch (TableAlreadyExistsException e) {
                 if (notExists) {
                     fail("Create Table should not fail");
                 }
             }

             // verify getTable rpcs
             verify(connectionQueryServices, sameClient ? never() : times(1)).getTable((PName)isNull(), eq(new byte[0]), eq(Bytes.toBytes(viewName)), anyLong(), anyLong());

             // verify no create table rpcs
             verify(connectionQueryServices, never()).createTable(anyListOf(Mutation.class),
                 any(byte[].class), any(PTableType.class), anyMap(), anyList(), any(byte[][].class),
                 eq(false), eq(false));
             reset(connectionQueryServices);

             // execute alter table ddl that adds the same column
             ddl = "ALTER VIEW " + viewName + " ADD " + (notExists ? "IF NOT EXISTS" : "") + " tagName varchar";
             try {
                 conn2.createStatement().execute(ddl);
                 if (!notExists) {
                     fail("Alter Table should fail");
                 }
             }
             catch (ColumnAlreadyExistsException e) {
                 if (notExists) {
                     fail("Alter Table should not fail");
                 }
             }

             // if not verify exists is true one call to add column table with empty mutation list (which does not make a rpc)
             // else verify no add column calls
             verify(connectionQueryServices, notExists ? times(1) : never() ).addColumn(eq(Collections.<Mutation>emptyList()), any(PTable.class), anyMap(), anySetOf(String.class), anyListOf(PColumn.class));

             // upsert one row
             conn2.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host2', 2.0)");
             conn2.commit();
             // verify data in base table
             ResultSet rs = conn2.createStatement().executeQuery("SELECT * from " + metricTableName);
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(1.0, rs.getDouble(2), 1e-6);
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(2.0, rs.getDouble(2), 1e-6);
             assertFalse(rs.next());
             // verify data in view
             rs = conn2.createStatement().executeQuery("SELECT * from " + viewName);
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(1.0, rs.getDouble(2), 1e-6);
             assertEquals("host1", rs.getString(3));
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(2.0, rs.getDouble(2), 1e-6);
             assertEquals("host2", rs.getString(3));
             assertFalse(rs.next());
         }
     }

     @Test
     public void testSameSchemaWithNotExistsSameClient() throws Exception {
         testTableWithSameSchema(true, true);
     }

     @Test
     public void testSameSchemaWithNotExistsDifferentClient() throws Exception {
         testTableWithSameSchema(true, false);
     }

     @Test
     public void testSameSchemaSameClient() throws Exception {
         testTableWithSameSchema(false, true);
     }

     @Test
     public void testSameSchemaDifferentClient() throws Exception {
         testTableWithSameSchema(false, false);
     }

     private void testAddColumns(boolean sameClient) throws Exception {
         Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
         try (Connection conn1 = DriverManager.getConnection(getUrl(), props);
                 Connection conn2 = sameClient ? conn1 : DriverManager.getConnection(getUrl(), props)) {

             String metricTableName = generateUniqueName();
             String viewName = generateUniqueName();
             String metricIdSeqTableName = generateUniqueName();

             // create sequence for auto partition
             conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + " CACHE 1");
             // create base table
             conn1.createStatement().execute("CREATE TABLE " + metricTableName + " (metricId INTEGER NOT NULL, metricVal1 DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
                     + " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
             // create view
             String ddl =
                     "CREATE VIEW IF NOT EXISTS "
                             + viewName + "( hostName varchar NOT NULL,"
                             + " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
                             + " AS SELECT * FROM " + metricTableName
                             + " UPDATE_CACHE_FREQUENCY=300000";
             conn1.createStatement().execute(ddl);

             conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal1) VALUES('host1', 1.0)");
             conn1.commit();

             // execute ddl that creates that same view with an additional pk column and regular column
             // and also changes the order of the pk columns (which is not respected since we only
             // allow appending columns)
             ddl =
                     "CREATE VIEW IF NOT EXISTS "
                             + viewName + "( instanceName varchar, hostName varchar, metricVal2 double, metricVal1 double"
                             + " CONSTRAINT HOSTNAME_PK PRIMARY KEY (instancename, hostName))"
                             + " AS SELECT * FROM " + metricTableName
                             + " UPDATE_CACHE_FREQUENCY=300000";
             conn2.createStatement().execute(ddl);

             conn2.createStatement().execute(
                 "UPSERT INTO " + viewName + "(hostName, instanceName, metricVal1, metricval2) VALUES('host2', 'instance2', 21.0, 22.0)");
             conn2.commit();

             conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal1) VALUES('host3', 3.0)");
             conn1.commit();

             // verify data exists
             ResultSet rs = conn2.createStatement().executeQuery("SELECT * from " + viewName);

             // verify the two columns were added correctly
             PTable table =
                     conn2.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, viewName));
             List<PColumn> pkColumns = table.getPKColumns();
             assertEquals(3,table.getPKColumns().size());
             // even though the second create view statement changed the order of the pk, the original order is maintained
             PColumn metricId = pkColumns.get(0);
             assertEquals("METRICID", metricId.getName().getString());
             assertFalse(metricId.isNullable());
             PColumn hostName = pkColumns.get(1);
             assertEquals("HOSTNAME", hostName.getName().getString());
             // hostname name is not nullable even though the second create statement changed it to nullable
             // since we only allow appending columns
             assertFalse(hostName.isNullable());
             PColumn instanceName = pkColumns.get(2);
             assertEquals("INSTANCENAME", instanceName.getName().getString());
             assertTrue(instanceName.isNullable());
             List<PColumn> columns = table.getColumns();
             assertEquals("METRICID", columns.get(0).getName().getString());
             assertEquals("METRICVAL1", columns.get(1).getName().getString());
             assertEquals("HOSTNAME", columns.get(2).getName().getString());
             assertEquals("INSTANCENAME", columns.get(3).getName().getString());
             assertEquals("METRICVAL2", columns.get(4).getName().getString());

             // verify the data
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(1.0, rs.getDouble(2), 1e-6);
             assertEquals("host1", rs.getString(3));
             assertEquals(null, rs.getString(4));
             assertEquals(0.0, rs.getDouble(5), 1e-6);
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(21.0, rs.getDouble(2), 1e-6);
             assertEquals("host2", rs.getString(3));
             assertEquals("instance2", rs.getString(4));
             assertEquals(22.0, rs.getDouble(5), 1e-6);
             assertTrue(rs.next());
             assertEquals(1, rs.getInt(1));
             assertEquals(3.0, rs.getDouble(2), 1e-6);
             assertEquals("host3", rs.getString(3));
             assertEquals(null, rs.getString(4));
             assertEquals(0.0, rs.getDouble(5), 1e-6);
             assertFalse(rs.next());
         }
     }

     @Test
     public void testAddColumnsSameClient() throws Exception {
         testAddColumns(true);
     }

     @Test
     public void testTableAddColumnsDifferentClient() throws Exception {
         testAddColumns(false);
     }

     @Test
     public void testValidateAttributes() throws Exception {
         Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
         try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
             String tableName = generateUniqueName();
             String viewName = generateUniqueName();
             try {
                 conn.createStatement().execute(
                     "create table IF NOT EXISTS " + tableName + " ( id char(1) NOT NULL,"
                             + " col1 integer NOT NULL,"
                             + " CONSTRAINT NAME_PK PRIMARY KEY (id, col1))"
                             + " APPEND_ONLY_SCHEMA = true");
                 fail("UPDATE_CACHE_FREQUENCY attribute must not be set to ALWAYS if APPEND_ONLY_SCHEMA is true");
             } catch (SQLException e) {
                 assertEquals(SQLExceptionCode.UPDATE_CACHE_FREQUENCY_INVALID.getErrorCode(),
                     e.getErrorCode());
             }

             conn.createStatement().execute(
                 "create table IF NOT EXISTS " + tableName + " ( id char(1) NOT NULL,"
                         + " col1 integer NOT NULL"
                         + " CONSTRAINT NAME_PK PRIMARY KEY (id, col1))"
                         + " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1000");
             conn.createStatement().execute(
                 "create view IF NOT EXISTS " + viewName + " (val1 integer) AS SELECT * FROM " + tableName);
             PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
             PTable view = pconn.getTable(new PTableKey(pconn.getTenantId(), viewName));
             assertEquals(true, view.isAppendOnlySchema());
             assertEquals(1000, view.getUpdateCacheFrequency());
         }
     }

     @Test
     public void testUpsertRowToDeletedTable() throws Exception {
         Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
         try (Connection conn1 = DriverManager.getConnection(getUrl(), props);
                 Connection conn2 = DriverManager.getConnection(getUrl(), props)) {
             String metricTableName = generateUniqueName();
             String viewName = generateUniqueName();
             String metricIdSeqTableName = generateUniqueName();
             // create sequence for auto partition
             conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + "  CACHE 1");
             // create base table
             conn1.createStatement().execute("CREATE TABLE " + metricTableName + " (metricId INTEGER NOT NULL, metricVal DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
                     + " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
             // create view
             String ddl =
                     "CREATE VIEW IF NOT EXISTS "
                             + viewName + "( hostName varchar NOT NULL,"
                             + " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
                             + " AS SELECT * FROM " + metricTableName
                             + " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=300000";
             conn1.createStatement().execute(ddl);

             // drop the table using a different connection
             conn2.createStatement().execute("DROP VIEW " + viewName);

             // upsert one row
             conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host1', 1.0)");
             // upsert doesn't fail since base table still exists
             conn1.commit();
         }
     }

 }
	/*
	* 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.phoenix.end2end;

	import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;
	import static org.junit.Assert.fail;
	import static org.mockito.Matchers.any;
	import static org.mockito.Matchers.anyList;
	import static org.mockito.Matchers.anyListOf;
	import static org.mockito.Matchers.anyLong;
	import static org.mockito.Matchers.anyMap;
	import static org.mockito.Matchers.anySetOf;
	import static org.mockito.Matchers.eq;
	import static org.mockito.Matchers.isNull;
	import static org.mockito.Mockito.never;
	import static org.mockito.Mockito.reset;
	import static org.mockito.Mockito.times;
	import static org.mockito.Mockito.verify;

	import java.sql.Connection;
	import java.sql.DriverManager;
	import java.sql.ResultSet;
	import java.sql.SQLException;
	import java.util.Collections;
	import java.util.List;
	import java.util.Properties;

	import org.apache.hadoop.hbase.client.Mutation;
	import org.apache.hadoop.hbase.util.Bytes;
	import org.apache.phoenix.exception.SQLExceptionCode;
	import org.apache.phoenix.jdbc.PhoenixConnection;
	import org.apache.phoenix.jdbc.PhoenixEmbeddedDriver;
	import org.apache.phoenix.query.ConnectionQueryServices;
	import org.apache.phoenix.schema.ColumnAlreadyExistsException;
	import org.apache.phoenix.schema.PColumn;
	import org.apache.phoenix.schema.PName;
	import org.apache.phoenix.schema.PTable;
	import org.apache.phoenix.schema.PTableKey;
	import org.apache.phoenix.schema.PTableType;
	import org.apache.phoenix.schema.TableAlreadyExistsException;
	import org.apache.phoenix.util.PropertiesUtil;
	import org.junit.Test;
	import org.mockito.Mockito;

	public class AppendOnlySchemaIT extends ParallelStatsDisabledIT {


	private void testTableWithSameSchema(boolean notExists, boolean sameClient) throws Exception {

	// use a spyed ConnectionQueryServices so we can verify calls to getTable
	ConnectionQueryServices connectionQueryServices =
	Mockito.spy(driver.getConnectionQueryServices(getUrl(),
	PropertiesUtil.deepCopy(TEST_PROPERTIES)));
	Properties props = new Properties();
	props.putAll(PhoenixEmbeddedDriver.DEFFAULT_PROPS.asMap());

	try (Connection conn1 = connectionQueryServices.connect(getUrl(), props);
	Connection conn2 = sameClient ? conn1 : connectionQueryServices.connect(getUrl(), props)) {

	String metricTableName = generateUniqueName();
	String viewName = generateUniqueName();
	String metricIdSeqTableName = generateUniqueName();
	// create sequence for auto partition
	conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + " CACHE 1");
	// create base table
	conn1.createStatement().execute("CREATE TABLE "+ metricTableName + "(metricId INTEGER NOT NULL, metricVal DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
	+ " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
	// create view
	String ddl =
	"CREATE VIEW " + (notExists ? "IF NOT EXISTS " : "")
	+ viewName + " ( hostName varchar NOT NULL, tagName varChar"
	+ " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
	+ " AS SELECT * FROM " + metricTableName
	+ " UPDATE_CACHE_FREQUENCY=300000";
	conn1.createStatement().execute(ddl);
	conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host1', 1.0)");
	conn1.commit();
	reset(connectionQueryServices);

	// execute same create ddl
	try {
	conn2.createStatement().execute(ddl);
	if (!notExists) {
	fail("Create Table should fail");
	}
	}
	catch (TableAlreadyExistsException e) {
	if (notExists) {
	fail("Create Table should not fail");
	}
	}

	// verify getTable rpcs
	verify(connectionQueryServices, sameClient ? never() : times(1)).getTable((PName)isNull(), eq(new byte[0]), eq(Bytes.toBytes(viewName)), anyLong(), anyLong());

	// verify no create table rpcs
	verify(connectionQueryServices, never()).createTable(anyListOf(Mutation.class),
	any(byte[].class), any(PTableType.class), anyMap(), anyList(), any(byte[][].class),
	eq(false), eq(false));
	reset(connectionQueryServices);

	// execute alter table ddl that adds the same column
	ddl = "ALTER VIEW " + viewName + " ADD " + (notExists ? "IF NOT EXISTS" : "") + " tagName varchar";
	try {
	conn2.createStatement().execute(ddl);
	if (!notExists) {
	fail("Alter Table should fail");
	}
	}
	catch (ColumnAlreadyExistsException e) {
	if (notExists) {
	fail("Alter Table should not fail");
	}
	}

	// if not verify exists is true one call to add column table with empty mutation list (which does not make a rpc)
	// else verify no add column calls
	verify(connectionQueryServices, notExists ? times(1) : never() ).addColumn(eq(Collections.<Mutation>emptyList()), any(PTable.class), anyMap(), anySetOf(String.class), anyListOf(PColumn.class));

	// upsert one row
	conn2.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host2', 2.0)");
	conn2.commit();
	// verify data in base table
	ResultSet rs = conn2.createStatement().executeQuery("SELECT * from " + metricTableName);
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(1.0, rs.getDouble(2), 1e-6);
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(2.0, rs.getDouble(2), 1e-6);
	assertFalse(rs.next());
	// verify data in view
	rs = conn2.createStatement().executeQuery("SELECT * from " + viewName);
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(1.0, rs.getDouble(2), 1e-6);
	assertEquals("host1", rs.getString(3));
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(2.0, rs.getDouble(2), 1e-6);
	assertEquals("host2", rs.getString(3));
	assertFalse(rs.next());
	}
	}

	@Test
	public void testSameSchemaWithNotExistsSameClient() throws Exception {
	testTableWithSameSchema(true, true);
	}

	@Test
	public void testSameSchemaWithNotExistsDifferentClient() throws Exception {
	testTableWithSameSchema(true, false);
	}

	@Test
	public void testSameSchemaSameClient() throws Exception {
	testTableWithSameSchema(false, true);
	}

	@Test
	public void testSameSchemaDifferentClient() throws Exception {
	testTableWithSameSchema(false, false);
	}

	private void testAddColumns(boolean sameClient) throws Exception {
	Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
	try (Connection conn1 = DriverManager.getConnection(getUrl(), props);
	Connection conn2 = sameClient ? conn1 : DriverManager.getConnection(getUrl(), props)) {

	String metricTableName = generateUniqueName();
	String viewName = generateUniqueName();
	String metricIdSeqTableName = generateUniqueName();

	// create sequence for auto partition
	conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + " CACHE 1");
	// create base table
	conn1.createStatement().execute("CREATE TABLE " + metricTableName + " (metricId INTEGER NOT NULL, metricVal1 DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
	+ " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
	// create view
	String ddl =
	"CREATE VIEW IF NOT EXISTS "
	+ viewName + "( hostName varchar NOT NULL,"
	+ " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
	+ " AS SELECT * FROM " + metricTableName
	+ " UPDATE_CACHE_FREQUENCY=300000";
	conn1.createStatement().execute(ddl);

	conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal1) VALUES('host1', 1.0)");
	conn1.commit();

	// execute ddl that creates that same view with an additional pk column and regular column
	// and also changes the order of the pk columns (which is not respected since we only
	// allow appending columns)
	ddl =
	"CREATE VIEW IF NOT EXISTS "
	+ viewName + "( instanceName varchar, hostName varchar, metricVal2 double, metricVal1 double"
	+ " CONSTRAINT HOSTNAME_PK PRIMARY KEY (instancename, hostName))"
	+ " AS SELECT * FROM " + metricTableName
	+ " UPDATE_CACHE_FREQUENCY=300000";
	conn2.createStatement().execute(ddl);

	conn2.createStatement().execute(
	"UPSERT INTO " + viewName + "(hostName, instanceName, metricVal1, metricval2) VALUES('host2', 'instance2', 21.0, 22.0)");
	conn2.commit();

	conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal1) VALUES('host3', 3.0)");
	conn1.commit();

	// verify data exists
	ResultSet rs = conn2.createStatement().executeQuery("SELECT * from " + viewName);

	// verify the two columns were added correctly
	PTable table =
	conn2.unwrap(PhoenixConnection.class).getTable(new PTableKey(null, viewName));
	List<PColumn> pkColumns = table.getPKColumns();
	assertEquals(3,table.getPKColumns().size());
	// even though the second create view statement changed the order of the pk, the original order is maintained
	PColumn metricId = pkColumns.get(0);
	assertEquals("METRICID", metricId.getName().getString());
	assertFalse(metricId.isNullable());
	PColumn hostName = pkColumns.get(1);
	assertEquals("HOSTNAME", hostName.getName().getString());
	// hostname name is not nullable even though the second create statement changed it to nullable
	// since we only allow appending columns
	assertFalse(hostName.isNullable());
	PColumn instanceName = pkColumns.get(2);
	assertEquals("INSTANCENAME", instanceName.getName().getString());
	assertTrue(instanceName.isNullable());
	List<PColumn> columns = table.getColumns();
	assertEquals("METRICID", columns.get(0).getName().getString());
	assertEquals("METRICVAL1", columns.get(1).getName().getString());
	assertEquals("HOSTNAME", columns.get(2).getName().getString());
	assertEquals("INSTANCENAME", columns.get(3).getName().getString());
	assertEquals("METRICVAL2", columns.get(4).getName().getString());

	// verify the data
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(1.0, rs.getDouble(2), 1e-6);
	assertEquals("host1", rs.getString(3));
	assertEquals(null, rs.getString(4));
	assertEquals(0.0, rs.getDouble(5), 1e-6);
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(21.0, rs.getDouble(2), 1e-6);
	assertEquals("host2", rs.getString(3));
	assertEquals("instance2", rs.getString(4));
	assertEquals(22.0, rs.getDouble(5), 1e-6);
	assertTrue(rs.next());
	assertEquals(1, rs.getInt(1));
	assertEquals(3.0, rs.getDouble(2), 1e-6);
	assertEquals("host3", rs.getString(3));
	assertEquals(null, rs.getString(4));
	assertEquals(0.0, rs.getDouble(5), 1e-6);
	assertFalse(rs.next());
	}
	}

	@Test
	public void testAddColumnsSameClient() throws Exception {
	testAddColumns(true);
	}

	@Test
	public void testTableAddColumnsDifferentClient() throws Exception {
	testAddColumns(false);
	}

	@Test
	public void testValidateAttributes() throws Exception {
	Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
	try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
	String tableName = generateUniqueName();
	String viewName = generateUniqueName();
	try {
	conn.createStatement().execute(
	"create table IF NOT EXISTS " + tableName + " ( id char(1) NOT NULL,"
	+ " col1 integer NOT NULL,"
	+ " CONSTRAINT NAME_PK PRIMARY KEY (id, col1))"
	+ " APPEND_ONLY_SCHEMA = true");
	fail("UPDATE_CACHE_FREQUENCY attribute must not be set to ALWAYS if APPEND_ONLY_SCHEMA is true");
	} catch (SQLException e) {
	assertEquals(SQLExceptionCode.UPDATE_CACHE_FREQUENCY_INVALID.getErrorCode(),
	e.getErrorCode());
	}

	conn.createStatement().execute(
	"create table IF NOT EXISTS " + tableName + " ( id char(1) NOT NULL,"
	+ " col1 integer NOT NULL"
	+ " CONSTRAINT NAME_PK PRIMARY KEY (id, col1))"
	+ " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1000");
	conn.createStatement().execute(
	"create view IF NOT EXISTS " + viewName + " (val1 integer) AS SELECT * FROM " + tableName);
	PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class);
	PTable view = pconn.getTable(new PTableKey(pconn.getTenantId(), viewName));
	assertEquals(true, view.isAppendOnlySchema());
	assertEquals(1000, view.getUpdateCacheFrequency());
	}
	}

	@Test
	public void testUpsertRowToDeletedTable() throws Exception {
	Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
	try (Connection conn1 = DriverManager.getConnection(getUrl(), props);
	Connection conn2 = DriverManager.getConnection(getUrl(), props)) {
	String metricTableName = generateUniqueName();
	String viewName = generateUniqueName();
	String metricIdSeqTableName = generateUniqueName();
	// create sequence for auto partition
	conn1.createStatement().execute("CREATE SEQUENCE " + metricIdSeqTableName + " CACHE 1");
	// create base table
	conn1.createStatement().execute("CREATE TABLE " + metricTableName + " (metricId INTEGER NOT NULL, metricVal DOUBLE, CONSTRAINT PK PRIMARY KEY(metricId))"
	+ " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=1, AUTO_PARTITION_SEQ=" + metricIdSeqTableName);
	// create view
	String ddl =
	"CREATE VIEW IF NOT EXISTS "
	+ viewName + "( hostName varchar NOT NULL,"
	+ " CONSTRAINT HOSTNAME_PK PRIMARY KEY (hostName))"
	+ " AS SELECT * FROM " + metricTableName
	+ " APPEND_ONLY_SCHEMA = true, UPDATE_CACHE_FREQUENCY=300000";
	conn1.createStatement().execute(ddl);

	// drop the table using a different connection
	conn2.createStatement().execute("DROP VIEW " + viewName);

	// upsert one row
	conn1.createStatement().execute("UPSERT INTO " + viewName + "(hostName, metricVal) VALUES('host1', 1.0)");
	// upsert doesn't fail since base table still exists
	conn1.commit();
	}
	}

	}