@Test public void testMigrateData() throws Exception { assertEquals("Table " + emptyTable + " is expected to be empty.", 0, executor.getTableRowCount(emptyTable)); assertEquals("Table " + table + " is expected to have " + NUMBER_OF_ROWS + " rows.", NUMBER_OF_ROWS, executor.getTableRowCount(table)); executor.migrateData(table, emptyTable); assertEquals("Table " + table + " is expected to be empty.", 0, executor.getTableRowCount(table)); assertEquals("Table " + emptyTable + " is expected to have " + NUMBER_OF_ROWS + " rows.", NUMBER_OF_ROWS, executor.getTableRowCount(emptyTable)); }
public void migrateData(String fromTable, String toTable) { String insertQuery = "INSERT INTO " + toTable + " SELECT * FROM " + fromTable; Statement stmt = null; Boolean oldAutoCommit = null; try { final long expectedInsertCount = getTableRowCount(fromTable); oldAutoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); stmt = connection.createStatement(); final int actualInsertCount = stmt.executeUpdate(insertQuery); if(expectedInsertCount == actualInsertCount) { LOG.info("Transferred " + actualInsertCount + " rows of staged data " + "from: " + fromTable + " to: " + toTable); connection.commit(); deleteTableData(fromTable); connection.commit(); } else { LOG.error("Rolling back as number of rows inserted into table: " + toTable + " was: " + actualInsertCount + " expected: " + expectedInsertCount); connection.rollback(); throw new SqoopException( GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018); } } catch(SQLException e) { logSQLException(e, "Got SQLException while migrating data from: " + fromTable + " to: " + toTable); throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018, e); } finally { if(stmt != null) { try { stmt.close(); } catch(SQLException e) { logSQLException(e, "Got SQLException at the time of closing statement."); } } if(oldAutoCommit != null) { try { connection.setAutoCommit(oldAutoCommit); } catch(SQLException e) { logSQLException(e, "Got SQLException while setting autoCommit mode."); } } } }
GenericJdbcExecutor { public void migrateData(String fromTable, String toTable) { String insertQuery = "INSERT INTO " + toTable + " SELECT * FROM " + fromTable; Statement stmt = null; Boolean oldAutoCommit = null; try { final long expectedInsertCount = getTableRowCount(fromTable); oldAutoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); stmt = connection.createStatement(); final int actualInsertCount = stmt.executeUpdate(insertQuery); if(expectedInsertCount == actualInsertCount) { LOG.info("Transferred " + actualInsertCount + " rows of staged data " + "from: " + fromTable + " to: " + toTable); connection.commit(); deleteTableData(fromTable); connection.commit(); } else { LOG.error("Rolling back as number of rows inserted into table: " + toTable + " was: " + actualInsertCount + " expected: " + expectedInsertCount); connection.rollback(); throw new SqoopException( GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018); } } catch(SQLException e) { logSQLException(e, "Got SQLException while migrating data from: " + fromTable + " to: " + toTable); throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018, e); } finally { if(stmt != null) { try { stmt.close(); } catch(SQLException e) { logSQLException(e, "Got SQLException at the time of closing statement."); } } if(oldAutoCommit != null) { try { connection.setAutoCommit(oldAutoCommit); } catch(SQLException e) { logSQLException(e, "Got SQLException while setting autoCommit mode."); } } } } }
GenericJdbcExecutor { public void migrateData(String fromTable, String toTable) { String insertQuery = "INSERT INTO " + toTable + " SELECT * FROM " + fromTable; Statement stmt = null; Boolean oldAutoCommit = null; try { final long expectedInsertCount = getTableRowCount(fromTable); oldAutoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); stmt = connection.createStatement(); final int actualInsertCount = stmt.executeUpdate(insertQuery); if(expectedInsertCount == actualInsertCount) { LOG.info("Transferred " + actualInsertCount + " rows of staged data " + "from: " + fromTable + " to: " + toTable); connection.commit(); deleteTableData(fromTable); connection.commit(); } else { LOG.error("Rolling back as number of rows inserted into table: " + toTable + " was: " + actualInsertCount + " expected: " + expectedInsertCount); connection.rollback(); throw new SqoopException( GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018); } } catch(SQLException e) { logSQLException(e, "Got SQLException while migrating data from: " + fromTable + " to: " + toTable); throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018, e); } finally { if(stmt != null) { try { stmt.close(); } catch(SQLException e) { logSQLException(e, "Got SQLException at the time of closing statement."); } } if(oldAutoCommit != null) { try { connection.setAutoCommit(oldAutoCommit); } catch(SQLException e) { logSQLException(e, "Got SQLException while setting autoCommit mode."); } } } } GenericJdbcExecutor(LinkConfig linkConfig); }
GenericJdbcExecutor { public void migrateData(String fromTable, String toTable) { String insertQuery = "INSERT INTO " + toTable + " SELECT * FROM " + fromTable; Statement stmt = null; Boolean oldAutoCommit = null; try { final long expectedInsertCount = getTableRowCount(fromTable); oldAutoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); stmt = connection.createStatement(); final int actualInsertCount = stmt.executeUpdate(insertQuery); if(expectedInsertCount == actualInsertCount) { LOG.info("Transferred " + actualInsertCount + " rows of staged data " + "from: " + fromTable + " to: " + toTable); connection.commit(); deleteTableData(fromTable); connection.commit(); } else { LOG.error("Rolling back as number of rows inserted into table: " + toTable + " was: " + actualInsertCount + " expected: " + expectedInsertCount); connection.rollback(); throw new SqoopException( GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018); } } catch(SQLException e) { logSQLException(e, "Got SQLException while migrating data from: " + fromTable + " to: " + toTable); throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018, e); } finally { if(stmt != null) { try { stmt.close(); } catch(SQLException e) { logSQLException(e, "Got SQLException at the time of closing statement."); } } if(oldAutoCommit != null) { try { connection.setAutoCommit(oldAutoCommit); } catch(SQLException e) { logSQLException(e, "Got SQLException while setting autoCommit mode."); } } } } GenericJdbcExecutor(LinkConfig linkConfig); ResultSet executeQuery(String sql); PreparedStatement createStatement(String sql); void setAutoCommit(boolean autoCommit); void deleteTableData(String tableName); void migrateData(String fromTable, String toTable); long getTableRowCount(String tableName); void executeUpdate(String sql); void beginBatch(String sql); void addBatch(Object[] array, Schema schema); void executeBatch(boolean commit); void endBatch(); String getPrimaryKey(String table); String[] getQueryColumns(String query); boolean existTable(String table); String qualify(String name, String qualifier); String[] dequalify(String name); String delimitIdentifier(String name); void close(); }
GenericJdbcExecutor { public void migrateData(String fromTable, String toTable) { String insertQuery = "INSERT INTO " + toTable + " SELECT * FROM " + fromTable; Statement stmt = null; Boolean oldAutoCommit = null; try { final long expectedInsertCount = getTableRowCount(fromTable); oldAutoCommit = connection.getAutoCommit(); connection.setAutoCommit(false); stmt = connection.createStatement(); final int actualInsertCount = stmt.executeUpdate(insertQuery); if(expectedInsertCount == actualInsertCount) { LOG.info("Transferred " + actualInsertCount + " rows of staged data " + "from: " + fromTable + " to: " + toTable); connection.commit(); deleteTableData(fromTable); connection.commit(); } else { LOG.error("Rolling back as number of rows inserted into table: " + toTable + " was: " + actualInsertCount + " expected: " + expectedInsertCount); connection.rollback(); throw new SqoopException( GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018); } } catch(SQLException e) { logSQLException(e, "Got SQLException while migrating data from: " + fromTable + " to: " + toTable); throw new SqoopException(GenericJdbcConnectorError.GENERIC_JDBC_CONNECTOR_0018, e); } finally { if(stmt != null) { try { stmt.close(); } catch(SQLException e) { logSQLException(e, "Got SQLException at the time of closing statement."); } } if(oldAutoCommit != null) { try { connection.setAutoCommit(oldAutoCommit); } catch(SQLException e) { logSQLException(e, "Got SQLException while setting autoCommit mode."); } } } } GenericJdbcExecutor(LinkConfig linkConfig); ResultSet executeQuery(String sql); PreparedStatement createStatement(String sql); void setAutoCommit(boolean autoCommit); void deleteTableData(String tableName); void migrateData(String fromTable, String toTable); long getTableRowCount(String tableName); void executeUpdate(String sql); void beginBatch(String sql); void addBatch(Object[] array, Schema schema); void executeBatch(boolean commit); void endBatch(); String getPrimaryKey(String table); String[] getQueryColumns(String query); boolean existTable(String table); String qualify(String name, String qualifier); String[] dequalify(String name); String delimitIdentifier(String name); void close(); static final String JDBC_PROPERTY_USERNAME; static final String JDBC_PROPERTY_PASSWORD; }