@Test public void testSchemaSqlChange() throws SQLException, ApiException, IOException { HiveClient hiveClient = Mockito.mock(HiveClient.class); List columnList= Lists.newArrayList(new ColumnDef("REC_UNIQUE_ID", "BIGINT",0), new ColumnDef("ORGNL_TRADE_DT", "DATE",1), new ColumnDef("MKT_CNTR_ID", "VARCHAR(1)",2), new ColumnDef("MBR_ID", "VARCHAR(4)",3), new ColumnDef("FIRM_MP_ID", "VARCHAR(4)",4), new ColumnDef("FIRM_CRD_NB", "INT",5), new ColumnDef("MBR_CLRG_NB", "VARCHAR(2)",6)); when(hiveClient.getExistingDDL(eq("metastor"), eq("test_managed_obj_test_orc"))) .thenReturn(new HiveTableSchema().toBuilder().columns(columnList).build()); when(hiveClient.tableExist(eq("metastor"), eq("test_managed_obj_test_orc"))).thenReturn(true); HiveHqlGenerator hiveHqlGenerator = new HiveHqlGenerator(); DataMgmtSvc dataMgmtSvc = Mockito.mock(DataMgmtSvc.class); hiveHqlGenerator.dataMgmtSvc = dataMgmtSvc; hiveHqlGenerator.hiveClient=hiveClient; hiveHqlGenerator.notificationSender=Mockito.mock(NotificationSender.class); JobDefinition jd = new JobDefinition(0L, "metastor", "test_managed_obj", "test", "orc", "2015-08-28", null, "aa", "trade_dt"); BusinessObjectFormat format = new BusinessObjectFormat(); format.setBusinessObjectFormatVersion(1); when(dataMgmtSvc.getDMFormat(eq(jd))).thenReturn(format); when(dataMgmtSvc.getTableSchema(eq(jd), eq(false))).thenReturn(DM_DDL); List<String> result = hiveHqlGenerator.schemaSql(true, jd); assertEquals("format cahnge", 2, result.size()); for(String s:result) { assertTrue("Contains Cascade", s.contains("CASCADE")); } }
public List<String> schemaSql(boolean schemaExists, JobDefinition jd) throws ApiException, SQLException { String tableName = jd.getTableName(); List<String> list = Lists.newArrayList(); if (schemaExists) { if(jd.getWfType() == ObjectProcessor.WF_TYPE_SINGLETON && jd.getPartitionKey().equalsIgnoreCase("partition")) { list.add(dataMgmtSvc.getTableSchema(jd, true)); } else { HiveTableSchema hiveTableSchema = hiveClient.getExistingDDL(jd.getObjectDefinition().getDbName(), tableName); BusinessObjectFormat format = dataMgmtSvc.getDMFormat(jd); String ddl = dataMgmtSvc.getTableSchema(jd, false); try { FormatChange change = detectSchemaChange(jd, hiveTableSchema, format, ddl); if (change.hasColumnChanges()) { boolean cascade = cascade(jd); String cascadeStr = ""; if(cascade) { cascadeStr = "CASCADE"; } if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); if(cascade) { sql = sql.substring(0, sql.lastIndexOf(";")) + " CASCADE;"; } list.add(sql); } else { for (Pair<ColumnDef, ColumnDef> pair : change.getNameChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } for (Pair<ColumnDef, ColumnDef> pair : change.getTypeChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } if (!change.getNewColumns().isEmpty()) { StringBuffer sb = new StringBuffer(); for (ColumnDef c : change.getNewColumns()) { sb.append(String.format("%s %s,", c.getName(), c.getType())); } sb.deleteCharAt(sb.length() - 1); list.add(String.format("Alter table %s add columns (%s) %s;", tableName, sb.toString(), cascadeStr)); } } } } catch(Exception ex) { log.warn("Error comparing formats", ex); notificationSender.sendNotificationEmail(ex.getMessage(), "Error comparing formats", jd); if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); list.add(sql); } } } } else { log.info("Table does not exist, create new " + jd.toString()); } log.info( "Table Schema: " + list ); return list; }
HiveHqlGenerator { public List<String> schemaSql(boolean schemaExists, JobDefinition jd) throws ApiException, SQLException { String tableName = jd.getTableName(); List<String> list = Lists.newArrayList(); if (schemaExists) { if(jd.getWfType() == ObjectProcessor.WF_TYPE_SINGLETON && jd.getPartitionKey().equalsIgnoreCase("partition")) { list.add(dataMgmtSvc.getTableSchema(jd, true)); } else { HiveTableSchema hiveTableSchema = hiveClient.getExistingDDL(jd.getObjectDefinition().getDbName(), tableName); BusinessObjectFormat format = dataMgmtSvc.getDMFormat(jd); String ddl = dataMgmtSvc.getTableSchema(jd, false); try { FormatChange change = detectSchemaChange(jd, hiveTableSchema, format, ddl); if (change.hasColumnChanges()) { boolean cascade = cascade(jd); String cascadeStr = ""; if(cascade) { cascadeStr = "CASCADE"; } if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); if(cascade) { sql = sql.substring(0, sql.lastIndexOf(";")) + " CASCADE;"; } list.add(sql); } else { for (Pair<ColumnDef, ColumnDef> pair : change.getNameChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } for (Pair<ColumnDef, ColumnDef> pair : change.getTypeChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } if (!change.getNewColumns().isEmpty()) { StringBuffer sb = new StringBuffer(); for (ColumnDef c : change.getNewColumns()) { sb.append(String.format("%s %s,", c.getName(), c.getType())); } sb.deleteCharAt(sb.length() - 1); list.add(String.format("Alter table %s add columns (%s) %s;", tableName, sb.toString(), cascadeStr)); } } } } catch(Exception ex) { log.warn("Error comparing formats", ex); notificationSender.sendNotificationEmail(ex.getMessage(), "Error comparing formats", jd); if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); list.add(sql); } } } } else { log.info("Table does not exist, create new " + jd.toString()); } log.info( "Table Schema: " + list ); return list; } }
HiveHqlGenerator { public List<String> schemaSql(boolean schemaExists, JobDefinition jd) throws ApiException, SQLException { String tableName = jd.getTableName(); List<String> list = Lists.newArrayList(); if (schemaExists) { if(jd.getWfType() == ObjectProcessor.WF_TYPE_SINGLETON && jd.getPartitionKey().equalsIgnoreCase("partition")) { list.add(dataMgmtSvc.getTableSchema(jd, true)); } else { HiveTableSchema hiveTableSchema = hiveClient.getExistingDDL(jd.getObjectDefinition().getDbName(), tableName); BusinessObjectFormat format = dataMgmtSvc.getDMFormat(jd); String ddl = dataMgmtSvc.getTableSchema(jd, false); try { FormatChange change = detectSchemaChange(jd, hiveTableSchema, format, ddl); if (change.hasColumnChanges()) { boolean cascade = cascade(jd); String cascadeStr = ""; if(cascade) { cascadeStr = "CASCADE"; } if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); if(cascade) { sql = sql.substring(0, sql.lastIndexOf(";")) + " CASCADE;"; } list.add(sql); } else { for (Pair<ColumnDef, ColumnDef> pair : change.getNameChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } for (Pair<ColumnDef, ColumnDef> pair : change.getTypeChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } if (!change.getNewColumns().isEmpty()) { StringBuffer sb = new StringBuffer(); for (ColumnDef c : change.getNewColumns()) { sb.append(String.format("%s %s,", c.getName(), c.getType())); } sb.deleteCharAt(sb.length() - 1); list.add(String.format("Alter table %s add columns (%s) %s;", tableName, sb.toString(), cascadeStr)); } } } } catch(Exception ex) { log.warn("Error comparing formats", ex); notificationSender.sendNotificationEmail(ex.getMessage(), "Error comparing formats", jd); if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); list.add(sql); } } } } else { log.info("Table does not exist, create new " + jd.toString()); } log.info( "Table Schema: " + list ); return list; }  }
HiveHqlGenerator { public List<String> schemaSql(boolean schemaExists, JobDefinition jd) throws ApiException, SQLException { String tableName = jd.getTableName(); List<String> list = Lists.newArrayList(); if (schemaExists) { if(jd.getWfType() == ObjectProcessor.WF_TYPE_SINGLETON && jd.getPartitionKey().equalsIgnoreCase("partition")) { list.add(dataMgmtSvc.getTableSchema(jd, true)); } else { HiveTableSchema hiveTableSchema = hiveClient.getExistingDDL(jd.getObjectDefinition().getDbName(), tableName); BusinessObjectFormat format = dataMgmtSvc.getDMFormat(jd); String ddl = dataMgmtSvc.getTableSchema(jd, false); try { FormatChange change = detectSchemaChange(jd, hiveTableSchema, format, ddl); if (change.hasColumnChanges()) { boolean cascade = cascade(jd); String cascadeStr = ""; if(cascade) { cascadeStr = "CASCADE"; } if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); if(cascade) { sql = sql.substring(0, sql.lastIndexOf(";")) + " CASCADE;"; } list.add(sql); } else { for (Pair<ColumnDef, ColumnDef> pair : change.getNameChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } for (Pair<ColumnDef, ColumnDef> pair : change.getTypeChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } if (!change.getNewColumns().isEmpty()) { StringBuffer sb = new StringBuffer(); for (ColumnDef c : change.getNewColumns()) { sb.append(String.format("%s %s,", c.getName(), c.getType())); } sb.deleteCharAt(sb.length() - 1); list.add(String.format("Alter table %s add columns (%s) %s;", tableName, sb.toString(), cascadeStr)); } } } } catch(Exception ex) { log.warn("Error comparing formats", ex); notificationSender.sendNotificationEmail(ex.getMessage(), "Error comparing formats", jd); if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); list.add(sql); } } } } else { log.info("Table does not exist, create new " + jd.toString()); } log.info( "Table Schema: " + list ); return list; }  List<String> schemaSql(boolean schemaExists, JobDefinition jd); String buildHql(JobDefinition jd, List<String> partitions); }
HiveHqlGenerator { public List<String> schemaSql(boolean schemaExists, JobDefinition jd) throws ApiException, SQLException { String tableName = jd.getTableName(); List<String> list = Lists.newArrayList(); if (schemaExists) { if(jd.getWfType() == ObjectProcessor.WF_TYPE_SINGLETON && jd.getPartitionKey().equalsIgnoreCase("partition")) { list.add(dataMgmtSvc.getTableSchema(jd, true)); } else { HiveTableSchema hiveTableSchema = hiveClient.getExistingDDL(jd.getObjectDefinition().getDbName(), tableName); BusinessObjectFormat format = dataMgmtSvc.getDMFormat(jd); String ddl = dataMgmtSvc.getTableSchema(jd, false); try { FormatChange change = detectSchemaChange(jd, hiveTableSchema, format, ddl); if (change.hasColumnChanges()) { boolean cascade = cascade(jd); String cascadeStr = ""; if(cascade) { cascadeStr = "CASCADE"; } if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); if(cascade) { sql = sql.substring(0, sql.lastIndexOf(";")) + " CASCADE;"; } list.add(sql); } else { for (Pair<ColumnDef, ColumnDef> pair : change.getNameChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } for (Pair<ColumnDef, ColumnDef> pair : change.getTypeChanges()) { ColumnDef existing = pair.getFirst(); ColumnDef newColum = pair.getSecond(); list.add(String.format("Alter table %s change %s %s %s %s;", tableName, existing.getName(), newColum.getName(), newColum.getType(), cascadeStr)); } if (!change.getNewColumns().isEmpty()) { StringBuffer sb = new StringBuffer(); for (ColumnDef c : change.getNewColumns()) { sb.append(String.format("%s %s,", c.getName(), c.getType())); } sb.deleteCharAt(sb.length() - 1); list.add(String.format("Alter table %s add columns (%s) %s;", tableName, sb.toString(), cascadeStr)); } } } } catch(Exception ex) { log.warn("Error comparing formats", ex); notificationSender.sendNotificationEmail(ex.getMessage(), "Error comparing formats", jd); if (!jd.getObjectDefinition().getFileType().equalsIgnoreCase("ORC")) { String sql = dataMgmtSvc.getTableSchema(jd, true); list.add(sql); } } } } else { log.info("Table does not exist, create new " + jd.toString()); } log.info( "Table Schema: " + list ); return list; }  List<String> schemaSql(boolean schemaExists, JobDefinition jd); String buildHql(JobDefinition jd, List<String> partitions); static final String SUBMITTED_BY_JOB_PROCESSOR; }