@Test public void testRidiculousRowCount() throws ClassNotFoundException, SQLException, InitializationException, IOException { long rowCount = Long.parseLong(Integer.toString(Integer.MAX_VALUE)) + 100; int partitionSize = 1000000; int expectedFileCount = (int) (rowCount / partitionSize) + 1; Connection conn = mock(Connection.class); when(dbcp.getConnection()).thenReturn(conn); Statement st = mock(Statement.class); when(conn.createStatement()).thenReturn(st); doNothing().when(st).close(); ResultSet rs = mock(ResultSet.class); when(st.executeQuery(anyString())).thenReturn(rs); when(rs.next()).thenReturn(true); when(rs.getInt(1)).thenReturn((int) rowCount); when(rs.getLong(1)).thenReturn(rowCount); final ResultSetMetaData resultSetMetaData = mock(ResultSetMetaData.class); when(rs.getMetaData()).thenReturn(resultSetMetaData); when(resultSetMetaData.getColumnCount()).thenReturn(2); when(resultSetMetaData.getTableName(1)).thenReturn(""); when(resultSetMetaData.getColumnType(1)).thenReturn(Types.INTEGER); when(resultSetMetaData.getColumnName(1)).thenReturn("COUNT"); when(resultSetMetaData.getColumnType(2)).thenReturn(Types.INTEGER); when(resultSetMetaData.getColumnName(2)).thenReturn("ID"); when(rs.getInt(2)).thenReturn(1000); runner.setProperty(GenerateTableFetch.TABLE_NAME, "TEST_QUERY_DB_TABLE"); runner.setIncomingConnection(false); runner.setProperty(GenerateTableFetch.MAX_VALUE_COLUMN_NAMES, "ID"); runner.setProperty(GenerateTableFetch.PARTITION_SIZE, Integer.toString(partitionSize)); runner.run(); runner.assertAllFlowFilesTransferred(REL_SUCCESS, expectedFileCount); MockFlowFile flowFile = runner.getFlowFilesForRelationship(REL_SUCCESS).get(0); String query = new String(flowFile.toByteArray()); assertEquals("SELECT * FROM TEST_QUERY_DB_TABLE WHERE 1=1 ORDER BY ID FETCH NEXT 1000000 ROWS ONLY", query); runner.clearTransferState(); }
private Integer getColumnType(String tableName, String colName, DatabaseAdapter adapter) { final String fullyQualifiedStateKey = getStateKey(tableName, colName, adapter); Integer type = columnTypeMap.get(fullyQualifiedStateKey); if (type == null && !isDynamicTableName) { type = columnTypeMap.get(getStateKey(null, colName, adapter)); } return type; }
GenerateTableFetch extends AbstractDatabaseFetchProcessor { private Integer getColumnType(String tableName, String colName, DatabaseAdapter adapter) { final String fullyQualifiedStateKey = getStateKey(tableName, colName, adapter); Integer type = columnTypeMap.get(fullyQualifiedStateKey); if (type == null && !isDynamicTableName) { type = columnTypeMap.get(getStateKey(null, colName, adapter)); } return type; } }
GenerateTableFetch extends AbstractDatabaseFetchProcessor { private Integer getColumnType(String tableName, String colName, DatabaseAdapter adapter) { final String fullyQualifiedStateKey = getStateKey(tableName, colName, adapter); Integer type = columnTypeMap.get(fullyQualifiedStateKey); if (type == null && !isDynamicTableName) { type = columnTypeMap.get(getStateKey(null, colName, adapter)); } return type; } GenerateTableFetch(); }
GenerateTableFetch extends AbstractDatabaseFetchProcessor { private Integer getColumnType(String tableName, String colName, DatabaseAdapter adapter) { final String fullyQualifiedStateKey = getStateKey(tableName, colName, adapter); Integer type = columnTypeMap.get(fullyQualifiedStateKey); if (type == null && !isDynamicTableName) { type = columnTypeMap.get(getStateKey(null, colName, adapter)); } return type; } GenerateTableFetch(); @Override Set<Relationship> getRelationships(); @Override @OnScheduled void setup(final ProcessContext context); @OnStopped void stop(); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); }
GenerateTableFetch extends AbstractDatabaseFetchProcessor { private Integer getColumnType(String tableName, String colName, DatabaseAdapter adapter) { final String fullyQualifiedStateKey = getStateKey(tableName, colName, adapter); Integer type = columnTypeMap.get(fullyQualifiedStateKey); if (type == null && !isDynamicTableName) { type = columnTypeMap.get(getStateKey(null, colName, adapter)); } return type; } GenerateTableFetch(); @Override Set<Relationship> getRelationships(); @Override @OnScheduled void setup(final ProcessContext context); @OnStopped void stop(); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); static final PropertyDescriptor PARTITION_SIZE; static final Relationship REL_FAILURE; }