/*
 * Licensed 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 com.facebook.presto.raptor.metadata;

import com.facebook.airlift.log.Logger;
import io.airlift.units.Duration;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.IDBI;
import org.skife.jdbi.v2.exceptions.UnableToObtainConnectionException;

import java.util.concurrent.TimeUnit;

public final class SchemaDaoUtil
{
    private static final Logger log = Logger.get(SchemaDaoUtil.class);

    private SchemaDaoUtil() {}

    public static void createTablesWithRetry(IDBI dbi)
    {
        Duration delay = new Duration(2, TimeUnit.SECONDS);
        while (true) {
            try (Handle handle = dbi.open()) {
                createTables(handle.attach(SchemaDao.class));
                return;
            }
            catch (UnableToObtainConnectionException e) {
                log.warn("Failed to connect to database. Will retry again in %s. Exception: %s", delay, e.getMessage());
                sleep(delay);
            }
        }
    }

    private static void createTables(SchemaDao dao)
    {
        dao.createTableDistributions();
        dao.createTableTables();
        dao.createTableColumns();
        dao.createTableViews();
        dao.createTableNodes();
        dao.createTableShards();
        dao.createTableShardNodes();
        dao.createTableExternalBatches();
        dao.createTableTransactions();
        dao.createTableCreatedShards();
        dao.createTableDeletedShards();
        dao.createTableBuckets();
        dao.createTableShardOrganizerJobs();

        // for upgrading compatibility
        alterTable(dao::alterTableTablesWithDeltaDelete);
        alterTable(dao::alterTableTablesWithDeltaCount);
        alterTable(dao::alterTableShardsWithIsDelta);
        alterTable(dao::alterTableShardsWithDeltaUuid);
    }

    private static void alterTable(Runnable alterTableFunction)
    {
        try {
            alterTableFunction.run();
        }
        catch (Exception e) {
            log.warn("Alter table failed: %s", e.getMessage());
        }
    }

    private static void sleep(Duration duration)
    {
        try {
            Thread.sleep(duration.toMillis());
        }
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
}

@@ -811,16 +811,202 @@
 ption;%0A%0A
+import java.sql.Connection;%0Aimport java.sql.ResultSet;%0Aimport java.sql.ResultSetMetaData;%0Aimport java.sql.SQLException;%0Aimport java.sql.SQLTransientException;%0Aimport java.sql.Statement;%0A
 import j
@@ -1424,24 +1424,109 @@
 ao.class));%0A
+                alterTables(handle.getConnection(), handle.attach(SchemaDao.class));%0A
             
@@ -1596,32 +1596,56 @@
 ectionException 
+%7C SQLTransientException 
 e) %7B%0A           
@@ -1790,32 +1790,130 @@
 ;%0A            %7D%0A
+            catch (SQLException e) %7B%0A                throw new RuntimeException(e);%0A            %7D%0A
         %7D%0A    %7D%0A
@@ -2452,16 +2452,134 @@
 Jobs();%0A
+    %7D%0A%0A    private static void alterTables(Connection connection, SchemaDao dao)%0A            throws SQLException%0A    %7B
 %0A       
@@ -2665,16 +2665,69 @@
 taDelete
+, %22tables%22, %22table_supports_delta_delete%22, connection
 );%0A     
@@ -2775,16 +2775,53 @@
 ltaCount
+, %22tables%22, %22delta_count%22, connection
 );%0A     
@@ -2866,16 +2866,50 @@
 hIsDelta
+, %22shards%22, %22is_delta%22, connection
 );%0A     
@@ -2956,16 +2956,52 @@
 eltaUuid
+, %22shards%22, %22delta_uuid%22, connection
 );%0A    %7D
@@ -3064,37 +3064,259 @@
 Function
-)%0A    %7B%0A        try %7B
+, String tableName, String columnName, Connection connection)%0A            throws SQLException%0A    %7B%0A        if (!findColumnName(tableName, columnName, connection)) %7B%0A            log.info(%22Alter table %25s, add column %25s%22, tableName, columnName);
 %0A       
@@ -3360,27 +3360,143 @@
   %7D%0A    
+%7D%0A%0A
     
-catch (
+private static boolean findColumnName(String tableName, String columnName, Connection connection)%0A            throws SQL
 Exceptio
@@ -3488,35 +3488,36 @@
 ows SQLException
- e)
+%0A   
  %7B%0A            l
@@ -3515,73 +3515,468 @@
     
-    log.warn(%22Alter table failed: %25s%22, e.getMessage());%0A        %7D
+Statement statement = connection.createStatement();%0A        ResultSet results = statement.executeQuery(%22SELECT * FROM %22 + tableName + %22 LIMIT 0%22);%0A        ResultSetMetaData metadata = results.getMetaData();%0A        int columnCount = metadata.getColumnCount();%0A        for (int i = 0; i %3C columnCount; i++) %7B%0A            if (columnName.equalsIgnoreCase(metadata.getColumnName(i + 1))) %7B%0A                return true;%0A            %7D%0A        %7D%0A        return false;
 %0A   

Check column exist before alter table for raptor delta delete