@Test public void testPartitionColChange() throws Exception { HiveHqlGenerator hiveHqlGenerator = new HiveHqlGenerator(); HiveTableSchema hiveTableSchema = HiveClientTest.getTestHiveTableSchema(); String ddl = "CREATE EXTERNAL TABLE IF NOT EXISTS `string` (\n" + " `TDATE` DATE,\n" + " `ACTIVITY_NAME` VARCHAR(35),\n" + " `TRANSACTN_DT_KEY` DECIMAL(10),\n" + " `OPT_PROD_KEY` DECIMAL(21),\n" + " `OPT_SYMBOL` VARCHAR(10),\n" + " `OPT_EXP_YR_MTH` DECIMAL(20),\n" + " `STRK_PRC` DECIMAL(20,8),\n" + " `PUT_CALL_CLASS` CHAR(1),\n" + " `CM` VARCHAR(5),\n" + " `FIRM` VARCHAR(100),\n" + " `CRD` VARCHAR(8),\n" + " `EXCEPTION_CAT` VARCHAR(15),\n" + " `OCC_POS_TRNSFR_ADJ_ID` VARCHAR(10),\n" + " `REC_ID` DECIMAL(10),\n" + " `CM_OPEN_POS` DECIMAL(20),\n" + " `OPTION_OPEN_POS` DECIMAL(20),\n" + " `XCPTN_ID` BIGINT)\n" + "PARTITIONED BY (`begindate` DATE, `test` DECIMAL(20,8))\n" + "ROW FORMAT DELIMITED FIELDS TERMINATED BY '\001' ESCAPED BY '\\' NULL DEFINED AS '\\N'\n" + "STORED AS TEXTFILE;"; BusinessObjectFormat format = new BusinessObjectFormat(); format.setBusinessObjectFormatVersion(0); hiveHqlGenerator.notificationSender=Mockito.mock(NotificationSender.class); FormatChange change = hiveHqlGenerator.detectSchemaChange(new JobDefinition(1,"MRP","OPT_OCCADJ_OPENINT_DETAIL", "PRC","BZ", "","","",""), hiveTableSchema, format, ddl); TestCase.assertTrue("There is change", change.hasChange()); TestCase.assertTrue("There is change", change.isPartitonColumnChanged()); }
@VisibleForTesting FormatChange detectSchemaChange(JobDefinition jd, HiveTableSchema hiveTableSchema, BusinessObjectFormat format, String ddl) throws ApiException { List<Pair<ColumnDef, ColumnDef>> nameChanges = Lists.newArrayList(); List<Pair<ColumnDef, ColumnDef>> typeChanges = Lists.newArrayList(); List<ColumnDef> addedColumns = Lists.newArrayList(); HiveTableSchema newSchema = HiveClientImpl.getHiveTableSchema(ddl); List<ColumnDef> existingColumns = hiveTableSchema.getColumns(); List<ColumnDef> newColumns = newSchema.getColumns(); log.info("Existing columns = " + existingColumns.size() + ", ddl from Herd has columns = " + newColumns.size()); int minColumns = Math.min(existingColumns.size(), newColumns.size()); for (int i = 0; i < minColumns; i++) { ColumnDef existing = existingColumns.get(i); ColumnDef newColum = newColumns.get(i); if (!newColum.getName().equalsIgnoreCase(existing.getName())) { nameChanges.add(new Pair<>(existing, newColum)); } else if (!newColum.isSameType(existing)) { typeChanges.add(new Pair<>(existing, newColum)); } } if (newColumns.size() > existingColumns.size()) { for (int i = existingColumns.size(); i < newColumns.size(); i++) { ColumnDef newColum = newColumns.get(i); addedColumns.add(newColum); } } FormatChange change = FormatChange.builder().nameChanges(nameChanges).newColumns(addedColumns) .typeChanges(typeChanges).build(); List<ColumnDef> existingPartition = hiveTableSchema.getPartitionColumns(); List<ColumnDef> newPartition = newSchema.getPartitionColumns(); if(existingPartition.size() != newPartition.size()) { change.setPartitonColumnChanged(true); } else { for (int i = 0; i < existingPartition.size(); i++) { if(!existingPartition.get(i).isSameColumn(newPartition.get(i))) { change.setPartitonColumnChanged(true); } } } if (change.hasChange()) { notificationSender.sendFormatChangeEmail(change, format.getBusinessObjectFormatVersion(), jd, hiveTableSchema, newSchema); } return change; }
HiveHqlGenerator { @VisibleForTesting FormatChange detectSchemaChange(JobDefinition jd, HiveTableSchema hiveTableSchema, BusinessObjectFormat format, String ddl) throws ApiException { List<Pair<ColumnDef, ColumnDef>> nameChanges = Lists.newArrayList(); List<Pair<ColumnDef, ColumnDef>> typeChanges = Lists.newArrayList(); List<ColumnDef> addedColumns = Lists.newArrayList(); HiveTableSchema newSchema = HiveClientImpl.getHiveTableSchema(ddl); List<ColumnDef> existingColumns = hiveTableSchema.getColumns(); List<ColumnDef> newColumns = newSchema.getColumns(); log.info("Existing columns = " + existingColumns.size() + ", ddl from Herd has columns = " + newColumns.size()); int minColumns = Math.min(existingColumns.size(), newColumns.size()); for (int i = 0; i < minColumns; i++) { ColumnDef existing = existingColumns.get(i); ColumnDef newColum = newColumns.get(i); if (!newColum.getName().equalsIgnoreCase(existing.getName())) { nameChanges.add(new Pair<>(existing, newColum)); } else if (!newColum.isSameType(existing)) { typeChanges.add(new Pair<>(existing, newColum)); } } if (newColumns.size() > existingColumns.size()) { for (int i = existingColumns.size(); i < newColumns.size(); i++) { ColumnDef newColum = newColumns.get(i); addedColumns.add(newColum); } } FormatChange change = FormatChange.builder().nameChanges(nameChanges).newColumns(addedColumns) .typeChanges(typeChanges).build(); List<ColumnDef> existingPartition = hiveTableSchema.getPartitionColumns(); List<ColumnDef> newPartition = newSchema.getPartitionColumns(); if(existingPartition.size() != newPartition.size()) { change.setPartitonColumnChanged(true); } else { for (int i = 0; i < existingPartition.size(); i++) { if(!existingPartition.get(i).isSameColumn(newPartition.get(i))) { change.setPartitonColumnChanged(true); } } } if (change.hasChange()) { notificationSender.sendFormatChangeEmail(change, format.getBusinessObjectFormatVersion(), jd, hiveTableSchema, newSchema); } return change; } }
HiveHqlGenerator { @VisibleForTesting FormatChange detectSchemaChange(JobDefinition jd, HiveTableSchema hiveTableSchema, BusinessObjectFormat format, String ddl) throws ApiException { List<Pair<ColumnDef, ColumnDef>> nameChanges = Lists.newArrayList(); List<Pair<ColumnDef, ColumnDef>> typeChanges = Lists.newArrayList(); List<ColumnDef> addedColumns = Lists.newArrayList(); HiveTableSchema newSchema = HiveClientImpl.getHiveTableSchema(ddl); List<ColumnDef> existingColumns = hiveTableSchema.getColumns(); List<ColumnDef> newColumns = newSchema.getColumns(); log.info("Existing columns = " + existingColumns.size() + ", ddl from Herd has columns = " + newColumns.size()); int minColumns = Math.min(existingColumns.size(), newColumns.size()); for (int i = 0; i < minColumns; i++) { ColumnDef existing = existingColumns.get(i); ColumnDef newColum = newColumns.get(i); if (!newColum.getName().equalsIgnoreCase(existing.getName())) { nameChanges.add(new Pair<>(existing, newColum)); } else if (!newColum.isSameType(existing)) { typeChanges.add(new Pair<>(existing, newColum)); } } if (newColumns.size() > existingColumns.size()) { for (int i = existingColumns.size(); i < newColumns.size(); i++) { ColumnDef newColum = newColumns.get(i); addedColumns.add(newColum); } } FormatChange change = FormatChange.builder().nameChanges(nameChanges).newColumns(addedColumns) .typeChanges(typeChanges).build(); List<ColumnDef> existingPartition = hiveTableSchema.getPartitionColumns(); List<ColumnDef> newPartition = newSchema.getPartitionColumns(); if(existingPartition.size() != newPartition.size()) { change.setPartitonColumnChanged(true); } else { for (int i = 0; i < existingPartition.size(); i++) { if(!existingPartition.get(i).isSameColumn(newPartition.get(i))) { change.setPartitonColumnChanged(true); } } } if (change.hasChange()) { notificationSender.sendFormatChangeEmail(change, format.getBusinessObjectFormatVersion(), jd, hiveTableSchema, newSchema); } return change; }  }
HiveHqlGenerator { @VisibleForTesting FormatChange detectSchemaChange(JobDefinition jd, HiveTableSchema hiveTableSchema, BusinessObjectFormat format, String ddl) throws ApiException { List<Pair<ColumnDef, ColumnDef>> nameChanges = Lists.newArrayList(); List<Pair<ColumnDef, ColumnDef>> typeChanges = Lists.newArrayList(); List<ColumnDef> addedColumns = Lists.newArrayList(); HiveTableSchema newSchema = HiveClientImpl.getHiveTableSchema(ddl); List<ColumnDef> existingColumns = hiveTableSchema.getColumns(); List<ColumnDef> newColumns = newSchema.getColumns(); log.info("Existing columns = " + existingColumns.size() + ", ddl from Herd has columns = " + newColumns.size()); int minColumns = Math.min(existingColumns.size(), newColumns.size()); for (int i = 0; i < minColumns; i++) { ColumnDef existing = existingColumns.get(i); ColumnDef newColum = newColumns.get(i); if (!newColum.getName().equalsIgnoreCase(existing.getName())) { nameChanges.add(new Pair<>(existing, newColum)); } else if (!newColum.isSameType(existing)) { typeChanges.add(new Pair<>(existing, newColum)); } } if (newColumns.size() > existingColumns.size()) { for (int i = existingColumns.size(); i < newColumns.size(); i++) { ColumnDef newColum = newColumns.get(i); addedColumns.add(newColum); } } FormatChange change = FormatChange.builder().nameChanges(nameChanges).newColumns(addedColumns) .typeChanges(typeChanges).build(); List<ColumnDef> existingPartition = hiveTableSchema.getPartitionColumns(); List<ColumnDef> newPartition = newSchema.getPartitionColumns(); if(existingPartition.size() != newPartition.size()) { change.setPartitonColumnChanged(true); } else { for (int i = 0; i < existingPartition.size(); i++) { if(!existingPartition.get(i).isSameColumn(newPartition.get(i))) { change.setPartitonColumnChanged(true); } } } if (change.hasChange()) { notificationSender.sendFormatChangeEmail(change, format.getBusinessObjectFormatVersion(), jd, hiveTableSchema, newSchema); } return change; }  List<String> schemaSql(boolean schemaExists, JobDefinition jd); String buildHql(JobDefinition jd, List<String> partitions); }
HiveHqlGenerator { @VisibleForTesting FormatChange detectSchemaChange(JobDefinition jd, HiveTableSchema hiveTableSchema, BusinessObjectFormat format, String ddl) throws ApiException { List<Pair<ColumnDef, ColumnDef>> nameChanges = Lists.newArrayList(); List<Pair<ColumnDef, ColumnDef>> typeChanges = Lists.newArrayList(); List<ColumnDef> addedColumns = Lists.newArrayList(); HiveTableSchema newSchema = HiveClientImpl.getHiveTableSchema(ddl); List<ColumnDef> existingColumns = hiveTableSchema.getColumns(); List<ColumnDef> newColumns = newSchema.getColumns(); log.info("Existing columns = " + existingColumns.size() + ", ddl from Herd has columns = " + newColumns.size()); int minColumns = Math.min(existingColumns.size(), newColumns.size()); for (int i = 0; i < minColumns; i++) { ColumnDef existing = existingColumns.get(i); ColumnDef newColum = newColumns.get(i); if (!newColum.getName().equalsIgnoreCase(existing.getName())) { nameChanges.add(new Pair<>(existing, newColum)); } else if (!newColum.isSameType(existing)) { typeChanges.add(new Pair<>(existing, newColum)); } } if (newColumns.size() > existingColumns.size()) { for (int i = existingColumns.size(); i < newColumns.size(); i++) { ColumnDef newColum = newColumns.get(i); addedColumns.add(newColum); } } FormatChange change = FormatChange.builder().nameChanges(nameChanges).newColumns(addedColumns) .typeChanges(typeChanges).build(); List<ColumnDef> existingPartition = hiveTableSchema.getPartitionColumns(); List<ColumnDef> newPartition = newSchema.getPartitionColumns(); if(existingPartition.size() != newPartition.size()) { change.setPartitonColumnChanged(true); } else { for (int i = 0; i < existingPartition.size(); i++) { if(!existingPartition.get(i).isSameColumn(newPartition.get(i))) { change.setPartitonColumnChanged(true); } } } if (change.hasChange()) { notificationSender.sendFormatChangeEmail(change, format.getBusinessObjectFormatVersion(), jd, hiveTableSchema, newSchema); } return change; }  List<String> schemaSql(boolean schemaExists, JobDefinition jd); String buildHql(JobDefinition jd, List<String> partitions); static final String SUBMITTED_BY_JOB_PROCESSOR; }