/*
 * 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 com.google.common.base.Preconditions.checkArgument;
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_LOCKED;
import static org.apache.phoenix.query.ConnectionQueryServicesImpl.UPGRADE_MUTEX_UNLOCKED;
import static org.apache.phoenix.query.QueryConstants.BASE_TABLE_BASE_COLUMN_COUNT;
import static org.apache.phoenix.query.QueryConstants.DIVERGED_VIEW_BASE_COLUMN_COUNT;
import static org.apache.phoenix.util.UpgradeUtil.SELECT_BASE_COLUMN_COUNT_FROM_HEADER_ROW;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.hadoop.hbase.HConstants;
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.client.RowMutations;
import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.exception.UpgradeInProgressException;
import org.apache.phoenix.exception.UpgradeRequiredException;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.ConnectionQueryServicesImpl;
import org.apache.phoenix.query.DelegateConnectionQueryServices;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PNameFactory;
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.UpgradeUtil;
import org.junit.Before;
import org.junit.Test;

public class UpgradeIT extends ParallelStatsDisabledIT {

    private String tenantId;
    
    @Before
    public void generateTenantId() {
        tenantId = "T_" + generateUniqueName();
    }

    @Test
    public void testUpgradeForTenantViewWithSameColumnsAsBaseTable() throws Exception {
        String tableWithViewName = generateUniqueName();
        String viewTableName = generateUniqueName();
        testViewUpgrade(true, tenantId, null, tableWithViewName + "1", null, viewTableName + "1", ColumnDiff.EQUAL);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "", null, viewTableName + "2",
            ColumnDiff.EQUAL);
        testViewUpgrade(true, tenantId, null, tableWithViewName + "3", viewTableName + "SCHEMA", viewTableName + "3",
            ColumnDiff.EQUAL);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "4", viewTableName + "SCHEMA", viewTableName + "4",
            ColumnDiff.EQUAL);
        testViewUpgrade(true, tenantId, "SAMESCHEMA", tableWithViewName + "5", "SAMESCHEMA", viewTableName + "5",
            ColumnDiff.EQUAL);
    }

    @Test
    public void testUpgradeForTenantViewWithMoreColumnsThanBaseTable() throws Exception {
        String tableWithViewName = generateUniqueName();
        String viewTableName = generateUniqueName();
        testViewUpgrade(true, tenantId, null, tableWithViewName + "1", null, viewTableName + "1", ColumnDiff.MORE);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "", null, viewTableName + "2",
            ColumnDiff.MORE);
        testViewUpgrade(true, tenantId, null, tableWithViewName + "3", "VIEWSCHEMA", viewTableName + "3",
            ColumnDiff.MORE);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "4", "VIEWSCHEMA", viewTableName + "4",
            ColumnDiff.MORE);
        testViewUpgrade(true, tenantId, "SAMESCHEMA", tableWithViewName + "5", "SAMESCHEMA", viewTableName + "5",
            ColumnDiff.MORE);
    }

    @Test
    public void testUpgradeForViewWithSameColumnsAsBaseTable() throws Exception {
        String tableWithViewName = generateUniqueName();
        String viewTableName = generateUniqueName();
        testViewUpgrade(false, null, null, tableWithViewName + "1", null, viewTableName + "1", ColumnDiff.EQUAL);
        testViewUpgrade(false, null, "TABLESCHEMA", tableWithViewName + "", null, viewTableName + "2",
            ColumnDiff.EQUAL);
        testViewUpgrade(false, null, null, tableWithViewName + "3", "VIEWSCHEMA", viewTableName + "3",
            ColumnDiff.EQUAL);
        testViewUpgrade(false, null, "TABLESCHEMA", tableWithViewName + "4", "VIEWSCHEMA", viewTableName + "4",
            ColumnDiff.EQUAL);
        testViewUpgrade(false, null, "SAMESCHEMA", tableWithViewName + "5", "SAMESCHEMA", viewTableName + "5",
            ColumnDiff.EQUAL);
    }

    @Test
    public void testUpgradeForViewWithMoreColumnsThanBaseTable() throws Exception {
        String tableWithViewName = generateUniqueName();
        String viewTableName = generateUniqueName();
        testViewUpgrade(false, null, null, tableWithViewName + "1", null, viewTableName + "1", ColumnDiff.MORE);
        testViewUpgrade(false, null, "TABLESCHEMA", tableWithViewName + "", null, viewTableName + "2", ColumnDiff.MORE);
        testViewUpgrade(false, null, null, tableWithViewName + "3", "VIEWSCHEMA", viewTableName + "3", ColumnDiff.MORE);
        testViewUpgrade(false, null, "TABLESCHEMA", tableWithViewName + "4", "VIEWSCHEMA", viewTableName + "4",
            ColumnDiff.MORE);
        testViewUpgrade(false, null, "SAMESCHEMA", tableWithViewName + "5", "SAMESCHEMA", viewTableName + "5",
            ColumnDiff.MORE);
    }

    @Test
    public void testSettingBaseColumnCountWhenBaseTableColumnDropped() throws Exception {
        String tableWithViewName = generateUniqueName();
        String viewTableName = generateUniqueName();
        testViewUpgrade(true, tenantId, null, tableWithViewName + "1", null, viewTableName + "1", ColumnDiff.MORE);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "", null, viewTableName + "2",
            ColumnDiff.LESS);
        testViewUpgrade(true, tenantId, null, tableWithViewName + "3", "VIEWSCHEMA", viewTableName + "3",
            ColumnDiff.LESS);
        testViewUpgrade(true, tenantId, "TABLESCHEMA", tableWithViewName + "4", "VIEWSCHEMA", viewTableName + "4",
            ColumnDiff.LESS);
        testViewUpgrade(true, tenantId, "SAMESCHEMA", tableWithViewName + "5", "SAMESCHEMA", viewTableName + "5",
            ColumnDiff.LESS);
    }

    @Test
    public void testMapTableToNamespaceDuringUpgrade()
            throws SQLException, IOException, IllegalArgumentException, InterruptedException {
        String[] strings = new String[] { "a", "b", "c", "d" };

        try (Connection conn = DriverManager.getConnection(getUrl())) {
            String schemaName = "TEST";
            String phoenixFullTableName = schemaName + "." + generateUniqueName();
            String indexName = "IDX_" + generateUniqueName();
            String localIndexName = "LIDX_" + generateUniqueName();

            String viewName = "VIEW_" + generateUniqueName();
            String viewIndexName = "VIDX_" + generateUniqueName();

            String[] tableNames = new String[] { phoenixFullTableName, schemaName + "." + indexName,
                    schemaName + "." + localIndexName, "diff." + viewName, "test." + viewName, viewName};
            String[] viewIndexes = new String[] { "diff." + viewIndexName, "test." + viewIndexName };
            conn.createStatement().execute("CREATE TABLE " + phoenixFullTableName
                    + "(k VARCHAR PRIMARY KEY, v INTEGER, f INTEGER, g INTEGER NULL, h INTEGER NULL)");
            PreparedStatement upsertStmt = conn
                    .prepareStatement("UPSERT INTO " + phoenixFullTableName + " VALUES(?, ?, 0, 0, 0)");
            int i = 1;
            for (String str : strings) {
                upsertStmt.setString(1, str);
                upsertStmt.setInt(2, i++);
                upsertStmt.execute();
            }
            conn.commit();
            // creating local index
            conn.createStatement()
                    .execute("create local index " + localIndexName + " on " + phoenixFullTableName + "(K)");
            // creating global index
            conn.createStatement().execute("create index " + indexName + " on " + phoenixFullTableName + "(k)");
            // creating view in schema 'diff'
            conn.createStatement().execute("CREATE VIEW diff." + viewName + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // creating view in schema 'test'
            conn.createStatement().execute("CREATE VIEW test." + viewName + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            conn.createStatement().execute("CREATE VIEW " + viewName + "(col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // Creating index on views
            conn.createStatement().execute("create index " + viewIndexName + "  on diff." + viewName + "(col)");
            conn.createStatement().execute("create index " + viewIndexName + " on test." + viewName + "(col)");

            // validate data
            for (String tableName : tableNames) {
                ResultSet rs = conn.createStatement().executeQuery("select * from " + tableName);
                for (String str : strings) {
                    assertTrue(rs.next());
                    assertEquals(str, rs.getString(1));
                }
            }

            // validate view Index data
            for (String viewIndex : viewIndexes) {
                ResultSet rs = conn.createStatement().executeQuery("select * from " + viewIndex);
                for (String str : strings) {
                    assertTrue(rs.next());
                    assertEquals(str, rs.getString(2));
                }
            }

            HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
            assertTrue(admin.tableExists(phoenixFullTableName));
            assertTrue(admin.tableExists(schemaName + QueryConstants.NAME_SEPARATOR + indexName));
            assertTrue(admin.tableExists(MetaDataUtil.getViewIndexPhysicalName(Bytes.toBytes(phoenixFullTableName))));
            Properties props = new Properties();
            props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(true));
            props.setProperty(QueryServices.IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, Boolean.toString(false));
            admin.close();
            PhoenixConnection phxConn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);
            UpgradeUtil.upgradeTable(phxConn, phoenixFullTableName);
            UpgradeUtil.mapChildViewsToNamespace(phxConn, phoenixFullTableName,props);
            phxConn.close();
            props = new Properties();
            phxConn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);
            admin = phxConn.getQueryServices().getAdmin();
            String hbaseTableName = SchemaUtil.getPhysicalTableName(Bytes.toBytes(phoenixFullTableName), true)
                    .getNameAsString();
            assertTrue(admin.tableExists(hbaseTableName));
            assertTrue(admin.tableExists(Bytes.toBytes(hbaseTableName)));
            assertTrue(admin.tableExists(schemaName + QueryConstants.NAMESPACE_SEPARATOR + indexName));
            assertTrue(admin.tableExists(MetaDataUtil.getViewIndexPhysicalName(Bytes.toBytes(hbaseTableName))));
            i = 0;
            // validate data
            for (String tableName : tableNames) {
                ResultSet rs = phxConn.createStatement().executeQuery("select * from " + tableName);
                for (String str : strings) {
                    assertTrue(rs.next());
                    assertEquals(str, rs.getString(1));
                }
            }
            // validate view Index data
            for (String viewIndex : viewIndexes) {
                ResultSet rs = conn.createStatement().executeQuery("select * from " + viewIndex);
                for (String str : strings) {
                    assertTrue(rs.next());
                    assertEquals(str, rs.getString(2));
                }
            }
            PName tenantId = phxConn.getTenantId();
            PName physicalName = PNameFactory.newName(hbaseTableName);
            String oldSchemaName = MetaDataUtil.getViewIndexSequenceSchemaName(PNameFactory.newName(phoenixFullTableName),
                    false);
            String newSchemaName = MetaDataUtil.getViewIndexSequenceSchemaName(physicalName, true);
            String newSequenceName = MetaDataUtil.getViewIndexSequenceName(physicalName, tenantId, true);
            ResultSet rs = phxConn.createStatement()
                    .executeQuery("SELECT " + PhoenixDatabaseMetaData.CURRENT_VALUE + "  FROM "
                            + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE + " WHERE " + PhoenixDatabaseMetaData.TENANT_ID
                            + " IS NULL AND " + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + " = '" + newSchemaName
                            + "' AND " + PhoenixDatabaseMetaData.SEQUENCE_NAME + "='" + newSequenceName + "'");
            assertTrue(rs.next());
            assertEquals("-32765", rs.getString(1));
            rs = phxConn.createStatement().executeQuery("SELECT " + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + ","
                    + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + "," + PhoenixDatabaseMetaData.CURRENT_VALUE + "  FROM "
                    + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE + " WHERE " + PhoenixDatabaseMetaData.TENANT_ID
                    + " IS NULL AND " + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA + " = '" + oldSchemaName + "'");
            assertFalse(rs.next());
            phxConn.close();
            admin.close();
   
        }
    }

    @Test
    public void testMapMultiTenantTableToNamespaceDuringUpgrade() throws SQLException, SnapshotCreationException,
            IllegalArgumentException, IOException, InterruptedException {
        String[] strings = new String[] { "a", "b", "c", "d" };
        String schemaName1 = "S_" +generateUniqueName(); // TEST
        String schemaName2 = "S_" +generateUniqueName(); // DIFF
        String phoenixFullTableName = schemaName1 + "." + generateUniqueName();
        String hbaseTableName = SchemaUtil.getPhysicalTableName(Bytes.toBytes(phoenixFullTableName), true)
                .getNameAsString();
        String indexName = "IDX_" + generateUniqueName();
        String viewName = "V_" + generateUniqueName();
        String viewName1 = "V1_" + generateUniqueName();
        String viewIndexName = "V_IDX_" + generateUniqueName();
        String tenantViewIndexName = "V1_IDX_" + generateUniqueName();

        String[] tableNames = new String[] { phoenixFullTableName, schemaName2 + "." + viewName1, schemaName1 + "." + viewName1, viewName1 };
        String[] viewIndexes = new String[] { schemaName1 + "." + viewIndexName, schemaName2 + "." + viewIndexName };
        String[] tenantViewIndexes = new String[] { schemaName1 + "." + tenantViewIndexName, schemaName2 + "." + tenantViewIndexName };
        try (Connection conn = DriverManager.getConnection(getUrl())) {
            conn.createStatement().execute("CREATE TABLE " + phoenixFullTableName
                    + "(k VARCHAR not null, v INTEGER not null, f INTEGER, g INTEGER NULL, h INTEGER NULL CONSTRAINT pk PRIMARY KEY(k,v)) MULTI_TENANT=true");
            PreparedStatement upsertStmt = conn
                    .prepareStatement("UPSERT INTO " + phoenixFullTableName + " VALUES(?, ?, 0, 0, 0)");
            int i = 1;
            for (String str : strings) {
                upsertStmt.setString(1, str);
                upsertStmt.setInt(2, i++);
                upsertStmt.execute();
            }
            conn.commit();

            // creating global index
            conn.createStatement().execute("create index " + indexName + " on " + phoenixFullTableName + "(f)");
            // creating view in schema 'diff'
            conn.createStatement().execute("CREATE VIEW " + schemaName2 + "." + viewName + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // creating view in schema 'test'
            conn.createStatement().execute("CREATE VIEW " + schemaName1 + "." + viewName + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            conn.createStatement().execute("CREATE VIEW " + viewName + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // Creating index on views
            conn.createStatement().execute("create local index " + viewIndexName + " on " + schemaName2 + "." + viewName + "(col)");
            conn.createStatement().execute("create local index " + viewIndexName + " on " + schemaName1 + "." + viewName + "(col)");
        }
        Properties props = new Properties();
        String tenantId = strings[0];
        props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
        try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
            PreparedStatement upsertStmt = conn
                    .prepareStatement("UPSERT INTO " + phoenixFullTableName + "(k,v,f,g,h)  VALUES(?, ?, 0, 0, 0)");
            int i = 1;
            for (String str : strings) {
                upsertStmt.setString(1, str);
                upsertStmt.setInt(2, i++);
                upsertStmt.execute();
            }
            conn.commit();
            // creating view in schema 'diff'
            conn.createStatement()
                    .execute("CREATE VIEW " + schemaName2 + "." + viewName1 + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // creating view in schema 'test'
            conn.createStatement()
                    .execute("CREATE VIEW " + schemaName1 + "." + viewName1 + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            conn.createStatement().execute("CREATE VIEW " + viewName1 + " (col VARCHAR) AS SELECT * FROM " + phoenixFullTableName);
            // Creating index on views
            conn.createStatement().execute("create index " + tenantViewIndexName + " on " + schemaName2 + "." + viewName1 + "(col)");
            conn.createStatement().execute("create index " + tenantViewIndexName + " on " + schemaName1 + "." + viewName1 + "(col)");
        }

        props = new Properties();
        props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(true));
        props.setProperty(QueryServices.IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, Boolean.toString(false));
        PhoenixConnection phxConn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);
        UpgradeUtil.upgradeTable(phxConn, phoenixFullTableName);
        UpgradeUtil.mapChildViewsToNamespace(phxConn,phoenixFullTableName,props);
        props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
        phxConn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);
        int i = 1;
        String indexPhysicalTableName = Bytes
                .toString(MetaDataUtil.getViewIndexPhysicalName(Bytes.toBytes(hbaseTableName)));
        // validate data with tenant
        for (String tableName : tableNames) {
            assertTableUsed(phxConn, tableName, hbaseTableName);
            ResultSet rs = phxConn.createStatement().executeQuery("select * from " + tableName);
            assertTrue(rs.next());
            do {
                assertEquals(i++, rs.getInt(1));
            } while (rs.next());
            i = 1;
        }
        // validate view Index data
        for (String viewIndex : tenantViewIndexes) {
            assertTableUsed(phxConn, viewIndex, indexPhysicalTableName);
            ResultSet rs = phxConn.createStatement().executeQuery("select * from " + viewIndex);
            assertTrue(rs.next());
            do {
                assertEquals(i++, rs.getInt(2));
            } while (rs.next());
            i = 1;
        }
        phxConn.close();
        props.remove(PhoenixRuntime.TENANT_ID_ATTRIB);
        phxConn = DriverManager.getConnection(getUrl(), props).unwrap(PhoenixConnection.class);

        // validate view Index data
        for (String viewIndex : viewIndexes) {
            assertTableUsed(phxConn, viewIndex, hbaseTableName);
            ResultSet rs = phxConn.createStatement().executeQuery("select * from " + viewIndex);
            for (String str : strings) {
                assertTrue(rs.next());
                assertEquals(str, rs.getString(1));
            }
        }
        phxConn.close();
    }

    public void assertTableUsed(Connection conn, String phoenixTableName, String hbaseTableName) throws SQLException {
        ResultSet rs = conn.createStatement().executeQuery("EXPLAIN SELECT * FROM " + phoenixTableName);
        assertTrue(rs.next());
        assertTrue(rs.getString(1).contains(hbaseTableName));
    }
    

    @Test
    public void testSettingBaseColumnCountForMultipleViewsOnTable() throws Exception {
        String baseSchema = "S_" + generateUniqueName();
        String baseTable = "T_" + generateUniqueName();
        String fullBaseTableName = SchemaUtil.getTableName(baseSchema, baseTable);
        try (Connection conn = DriverManager.getConnection(getUrl())) {
            String baseTableDDL = "CREATE TABLE " + fullBaseTableName + " (TENANT_ID VARCHAR NOT NULL, PK1 VARCHAR NOT NULL, V1 INTEGER, V2 INTEGER CONSTRAINT NAME_PK PRIMARY KEY(TENANT_ID, PK1)) MULTI_TENANT = true";
            conn.createStatement().execute(baseTableDDL);

            String[] tenants = new String[] {"T_" + generateUniqueName(), "T_" + generateUniqueName()};
            Collections.sort(Arrays.asList(tenants));
            String[] tenantViews = new String[] {"V_" + generateUniqueName(), "V_" + generateUniqueName(), "V_" + generateUniqueName()};
            Collections.sort(Arrays.asList(tenantViews));
            String[] globalViews = new String[] {"G_" + generateUniqueName(), "G_" + generateUniqueName(), "G_" + generateUniqueName()};
            Collections.sort(Arrays.asList(globalViews));
            for (int i = 0; i < 2; i++) {
                // Create views for tenants;
                String tenant = tenants[i];
                try (Connection tenantConn = createTenantConnection(tenant)) {
                    String view = tenantViews[0];
                    // view with its own column
                    String viewDDL = "CREATE VIEW " + view + " AS SELECT * FROM " + fullBaseTableName;
                    tenantConn.createStatement().execute(viewDDL);
                    String addCols = "ALTER VIEW " + view + " ADD COL1 VARCHAR ";
                    tenantConn.createStatement().execute(addCols);
                    removeBaseColumnCountKV(tenant, null, view);

                    // view that has the last base table column removed
                    view = tenantViews[1];
                    viewDDL = "CREATE VIEW " + view + " AS SELECT * FROM " + fullBaseTableName;
                    tenantConn.createStatement().execute(viewDDL);
                    String droplastBaseCol = "ALTER VIEW " + view + " DROP COLUMN V2";
                    tenantConn.createStatement().execute(droplastBaseCol);
                    removeBaseColumnCountKV(tenant, null, view);

                    // view that has the middle base table column removed
                    view = tenantViews[2];
                    viewDDL = "CREATE VIEW " + view + " AS SELECT * FROM " + fullBaseTableName;
                    tenantConn.createStatement().execute(viewDDL);
                    String dropMiddileBaseCol = "ALTER VIEW " + view + " DROP COLUMN V1";
                    tenantConn.createStatement().execute(dropMiddileBaseCol);
                    removeBaseColumnCountKV(tenant, null, view);
                }
            }

            // create global views
            try (Connection globalConn = DriverManager.getConnection(getUrl())) {
                String globalView = globalViews[0];
                // view with its own column
                String viewDDL = "CREATE VIEW " + globalView + " AS SELECT * FROM " + fullBaseTableName;
                globalConn.createStatement().execute(viewDDL);
                String addCols = "ALTER VIEW " + globalView + " ADD COL1 VARCHAR ";
                globalConn.createStatement().execute(addCols);
                removeBaseColumnCountKV(null, null, globalView);

                // view that has the last base table column removed
                globalView = globalViews[1];
                viewDDL = "CREATE VIEW " + globalView + " AS SELECT * FROM " + fullBaseTableName;
                globalConn.createStatement().execute(viewDDL);
                String droplastBaseCol = "ALTER VIEW " + globalView + " DROP COLUMN V2";
                globalConn.createStatement().execute(droplastBaseCol);
                removeBaseColumnCountKV(null, null, globalView);

                // view that has the middle base table column removed
                globalView = globalViews[2];
                viewDDL = "CREATE VIEW " + globalView + " AS SELECT * FROM " + fullBaseTableName;
                globalConn.createStatement().execute(viewDDL);
                String dropMiddileBaseCol = "ALTER VIEW " + globalView + " DROP COLUMN V1";
                globalConn.createStatement().execute(dropMiddileBaseCol);
                removeBaseColumnCountKV(null, null, globalView);
            }
            
            // run upgrade
            UpgradeUtil.upgradeTo4_5_0(conn.unwrap(PhoenixConnection.class));
            
            // Verify base column counts for tenant specific views
            for (int i = 0; i < 2 ; i++) {
                String tenantId = tenants[i];
                checkBaseColumnCount(tenantId, null, tenantViews[0], 4);
                checkBaseColumnCount(tenantId, null, tenantViews[1], DIVERGED_VIEW_BASE_COLUMN_COUNT);
                checkBaseColumnCount(tenantId, null, tenantViews[2], DIVERGED_VIEW_BASE_COLUMN_COUNT);
            }
            
            // Verify base column count for global views
            checkBaseColumnCount(null, null, globalViews[0], 4);
            checkBaseColumnCount(null, null, globalViews[1], DIVERGED_VIEW_BASE_COLUMN_COUNT);
            checkBaseColumnCount(null, null, globalViews[2], DIVERGED_VIEW_BASE_COLUMN_COUNT);
        }
        
        
    }
    
    private enum ColumnDiff {
        MORE, EQUAL, LESS
    };

    private void testViewUpgrade(boolean tenantView, String tenantId, String baseTableSchema,
            String baseTableName, String viewSchema, String viewName, ColumnDiff diff)
            throws Exception {
        if (tenantView) {
            checkNotNull(tenantId);
        } else {
            checkArgument(tenantId == null);
        }
        Connection conn = DriverManager.getConnection(getUrl());
        String fullViewName = SchemaUtil.getTableName(viewSchema, viewName);
        String fullBaseTableName = SchemaUtil.getTableName(baseTableSchema, baseTableName);
        try {
            int expectedBaseColumnCount;
            conn.createStatement().execute(
                "CREATE TABLE IF NOT EXISTS " + fullBaseTableName + " ("
                        + " TENANT_ID CHAR(15) NOT NULL, " + " PK1 integer NOT NULL, "
                        + "PK2 bigint NOT NULL, " + "CF1.V1 VARCHAR, " + "CF2.V2 VARCHAR, "
                        + "V3 CHAR(100) ARRAY[4] "
                        + " CONSTRAINT NAME_PK PRIMARY KEY (TENANT_ID, PK1, PK2)"
                        + " ) MULTI_TENANT= true");
            
            // create a view with same columns as base table.
            try (Connection conn2 = getConnection(tenantView, tenantId)) {
                conn2.createStatement().execute(
                    "CREATE VIEW " + fullViewName + " AS SELECT * FROM " + fullBaseTableName);
            }

            if (diff == ColumnDiff.MORE) {
                    // add a column to the view
                    try (Connection conn3 = getConnection(tenantView, tenantId)) {
                        conn3.createStatement().execute(
                            "ALTER VIEW " + fullViewName + " ADD VIEW_COL1 VARCHAR");
                    }
            }
            if (diff == ColumnDiff.LESS) {
                try (Connection conn3 = getConnection(tenantView, tenantId)) {
                    conn3.createStatement().execute(
                        "ALTER VIEW " + fullViewName + " DROP COLUMN CF2.V2");
                }
                expectedBaseColumnCount = DIVERGED_VIEW_BASE_COLUMN_COUNT;
            } else {
                expectedBaseColumnCount = 6;
            }

            checkBaseColumnCount(tenantId, viewSchema, viewName, expectedBaseColumnCount);
            checkBaseColumnCount(null, baseTableSchema, baseTableName, BASE_TABLE_BASE_COLUMN_COUNT);
            
            // remove base column count kv so we can check whether the upgrade code is setting the 
            // base column count correctly.
            removeBaseColumnCountKV(tenantId, viewSchema, viewName);
            removeBaseColumnCountKV(null, baseTableSchema, baseTableName);

            // assert that the removing base column count key value worked correctly.
            checkBaseColumnCount(tenantId, viewSchema, viewName, 0);
            checkBaseColumnCount(null, baseTableSchema, baseTableName, 0);
            
            // run upgrade
            UpgradeUtil.upgradeTo4_5_0(conn.unwrap(PhoenixConnection.class));

            checkBaseColumnCount(tenantId, viewSchema, viewName, expectedBaseColumnCount);
            checkBaseColumnCount(null, baseTableSchema, baseTableName, BASE_TABLE_BASE_COLUMN_COUNT);
        } finally {
            conn.close();
        }
    }

    private static void checkBaseColumnCount(String tenantId, String schemaName, String tableName,
            int expectedBaseColumnCount) throws Exception {
        checkNotNull(tableName);
        Connection conn = DriverManager.getConnection(getUrl());
        String sql = SELECT_BASE_COLUMN_COUNT_FROM_HEADER_ROW;
        sql =
                String.format(sql, tenantId == null ? " IS NULL " : " = ? ",
                    schemaName == null ? "IS NULL" : " = ? ");
        int paramIndex = 1;
        PreparedStatement stmt = conn.prepareStatement(sql);
        if (tenantId != null) {
            stmt.setString(paramIndex++, tenantId);
        }
        if (schemaName != null) {
            stmt.setString(paramIndex++, schemaName);
        }
        stmt.setString(paramIndex, tableName);
        ResultSet rs = stmt.executeQuery();
        assertTrue(rs.next());
        assertEquals(expectedBaseColumnCount, rs.getInt(1));
        assertFalse(rs.next());
    }

    private static void
            removeBaseColumnCountKV(String tenantId, String schemaName, String tableName)
                    throws Exception {
        byte[] rowKey =
                SchemaUtil.getTableKey(tenantId == null ? new byte[0] : Bytes.toBytes(tenantId),
                    schemaName == null ? new byte[0] : Bytes.toBytes(schemaName),
                    Bytes.toBytes(tableName));
        Put viewColumnDefinitionPut = new Put(rowKey, HConstants.LATEST_TIMESTAMP);
        viewColumnDefinitionPut.add(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
            PhoenixDatabaseMetaData.BASE_COLUMN_COUNT_BYTES, HConstants.LATEST_TIMESTAMP, null);

        try (PhoenixConnection conn =
                (DriverManager.getConnection(getUrl())).unwrap(PhoenixConnection.class)) {
            try (HTableInterface htable =
                    conn.getQueryServices().getTable(
                        Bytes.toBytes(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME))) {
                RowMutations mutations = new RowMutations(rowKey);
                mutations.add(viewColumnDefinitionPut);
                htable.mutateRow(mutations);
            }
        }
    }
    
    @Test
    public void testUpgradeRequiredPreventsSQL() throws SQLException {
        String tableName = generateUniqueName();
        try (Connection conn = getConnection(false, null)) {
            conn.createStatement().execute(
                    "CREATE TABLE " + tableName
                            + " (PK1 VARCHAR NOT NULL, PK2 VARCHAR, KV1 VARCHAR, KV2 VARCHAR CONSTRAINT PK PRIMARY KEY(PK1, PK2))");
            final ConnectionQueryServices delegate = conn.unwrap(PhoenixConnection.class).getQueryServices();
            ConnectionQueryServices servicesWithUpgrade = new DelegateConnectionQueryServices(delegate) {
                @Override
                public boolean isUpgradeRequired() {
                    return true;
                }
            };
            try (PhoenixConnection phxConn = new PhoenixConnection(servicesWithUpgrade,
                    conn.unwrap(PhoenixConnection.class), HConstants.LATEST_TIMESTAMP)) {
                try {
                    phxConn.createStatement().execute(
                            "CREATE TABLE " + generateUniqueName()
                                    + " (k1 VARCHAR NOT NULL, k2 VARCHAR, CONSTRAINT PK PRIMARY KEY(K1,K2))");
                    fail("CREATE TABLE should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
                try {
                    phxConn.createStatement().execute("SELECT * FROM " + tableName);
                    fail("SELECT should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
                try {
                    phxConn.createStatement().execute("DELETE FROM " + tableName);
                    fail("DELETE should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
                try {
                    phxConn.createStatement().execute(
                            "CREATE INDEX " + tableName + "_IDX ON " + tableName + " (KV1) INCLUDE (KV2)" );
                    fail("CREATE INDEX should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
                try {
                    phxConn.createStatement().execute(
                            "UPSERT INTO " + tableName + " VALUES ('PK1', 'PK2', 'KV1', 'KV2')" );
                    fail("UPSERT VALUES should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
            }
        }
    }
    
    @Test
    public void testUpgradingConnectionBypassesUpgradeRequiredCheck() throws Exception {
        String tableName = generateUniqueName();
        try (Connection conn = getConnection(false, null)) {
            conn.createStatement()
                    .execute(
                            "CREATE TABLE "
                                    + tableName
                                    + " (PK1 VARCHAR NOT NULL, PK2 VARCHAR, KV1 VARCHAR, KV2 VARCHAR CONSTRAINT PK PRIMARY KEY(PK1, PK2))");
            final ConnectionQueryServices delegate = conn.unwrap(PhoenixConnection.class).getQueryServices();
            ConnectionQueryServices servicesWithUpgrade = new DelegateConnectionQueryServices(delegate) {
                @Override
                public boolean isUpgradeRequired() {
                    return true;
                }
            };
            try (PhoenixConnection phxConn = new PhoenixConnection(servicesWithUpgrade,
                    conn.unwrap(PhoenixConnection.class), HConstants.LATEST_TIMESTAMP)) {
                // Because upgrade is required, this SQL should fail.
                try {
                    phxConn.createStatement().executeQuery("SELECT * FROM " + tableName);
                    fail("SELECT should have failed with UpgradeRequiredException");
                } catch (UpgradeRequiredException expected) {

                }
                // Marking connection as the one running upgrade should let SQL execute fine.
                phxConn.setRunningUpgrade(true);
                phxConn.createStatement().execute(
                        "UPSERT INTO " + tableName + " VALUES ('PK1', 'PK2', 'KV1', 'KV2')" );
                phxConn.commit();
                try (ResultSet rs = phxConn.createStatement().executeQuery("SELECT * FROM " + tableName)) {
                    assertTrue(rs.next());
                    assertFalse(rs.next());
                }
            }
        }
    }
    
    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.add(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));
            try {
                ((ConnectionQueryServicesImpl)services)
                        .acquireUpgradeMutex(MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0, mutexRowKey);
                fail();
            } catch (UpgradeInProgressException expected) {

            }
            assertTrue(((ConnectionQueryServicesImpl)services).releaseUpgradeMutex(mutexRowKey));
            assertFalse(((ConnectionQueryServicesImpl)services).releaseUpgradeMutex(mutexRowKey));
        }
    }
    
    @Test
    public void testConcurrentUpgradeThrowsUprgadeInProgressException() throws Exception {
        final AtomicBoolean mutexStatus1 = new AtomicBoolean(false);
        final AtomicBoolean mutexStatus2 = new AtomicBoolean(false);
        final CountDownLatch latch = new CountDownLatch(2);
        final AtomicInteger numExceptions = new AtomicInteger(0);
        ConnectionQueryServices services = null;
        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);
            t1.setDaemon(true);
            Thread t2 = new Thread(task2);
            t2.setDaemon(true);
            t1.start();
            t2.start();
            latch.await();
            // make sure tasks didn't fail by calling get()
            task1.get();
            task2.get();
            assertTrue("One of the threads should have acquired the mutex", mutexStatus1.get() || mutexStatus2.get());
            assertNotEquals("One and only one thread should have acquired the mutex ", mutexStatus1.get(),
                    mutexStatus2.get());
            assertEquals("One and only one thread should have caught UpgradeRequiredException ", 1, numExceptions.get());
        }
    }
    
    private static class AcquireMutexRunnable implements Callable<Void> {
        
        private final AtomicBoolean acquireStatus;
        private final ConnectionQueryServices services;
        private final CountDownLatch latch;
        private final AtomicInteger numExceptions;
        private final byte[] mutexRowKey;
        public AcquireMutexRunnable(AtomicBoolean acquireStatus, ConnectionQueryServices services, CountDownLatch latch, AtomicInteger numExceptions, byte[] mutexKey) {
            this.acquireStatus = acquireStatus;
            this.services = services;
            this.latch = latch;
            this.numExceptions = numExceptions;
            this.mutexRowKey = mutexKey;
        }
        @Override
        public Void call() throws Exception {
            try {
                ((ConnectionQueryServicesImpl)services).acquireUpgradeMutex(
                        MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_7_0, mutexRowKey);
                acquireStatus.set(true);
            } catch (UpgradeInProgressException e) {
                numExceptions.incrementAndGet();
            } finally {
                latch.countDown();
            }
            return null;
        }
        
    }

    private Connection createTenantConnection(String tenantId) throws SQLException {
        Properties props = new Properties();
        props.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
        return DriverManager.getConnection(getUrl(), props);
    }

    private Connection getConnection(boolean tenantSpecific, String tenantId) throws SQLException {
        if (tenantSpecific) {
            checkNotNull(tenantId);
            return createTenantConnection(tenantId);
        }
        return DriverManager.getConnection(getUrl());
    }
    
}