// Check whether we have any special mappings that we need to maintain
Map<Integer, Integer> code_2_id = new HashMap<Integer, Integer>();
Map<Integer, Map<String, Long>> mapping_columns = new HashMap<Integer, Map<String, Long>>();
for (int col_code_idx = 0, cnt = columns.size(); col_code_idx < cnt; col_code_idx++) {
Column catalog_col = columns.get(col_code_idx);
String col_name = catalog_col.getName();
// Code Column -> Id Column Mapping
// Check to see whether this table has columns that we need to map their
// code values to tuple ids
String col_id_name = this.profile.code_columns.get(col_name);
if (col_id_name != null) {
Column catalog_id_col = catalog_tbl.getColumnByName(col_id_name);
assert(catalog_id_col != null) : "The id column " + catalog_tbl.getName() + "." + col_id_name + " is missing";
int col_id_idx = catalog_tbl.getColumnIndex(catalog_id_col);
code_2_id.put(col_code_idx, col_id_idx);
}
// Foreign Key Column to Code->Id Mapping
// If this columns references a foreign key that is used in the Code->Id mapping
// that we generating above, then we need to know when we should change the
// column value from a code to the id stored in our lookup table
if (this.profile.fkey_value_xref.containsKey(col_name)) {
String col_fkey_name = this.profile.fkey_value_xref.get(col_name);
mapping_columns.put(col_code_idx, this.profile.code_id_xref.get(col_fkey_name));
}
} // FOR
int row_idx = 0;
int row_batch = 0;
try {
String insert_sql = SQLUtil.getInsertSQL(catalog_tbl);
PreparedStatement insert_stmt = this.conn.prepareStatement(insert_sql);
int sqlTypes[] = catalog_tbl.getColumnTypes();
for (Object tuple[] : iterable) {
assert(tuple[0] != null) : "The primary key for " + catalog_tbl.getName() + " is null";
// AIRPORT
if (is_airport) {
// Skip any airport that does not have flights
int col_code_idx = catalog_tbl.getColumnByName("AP_CODE").getIndex();
if (profile.hasFlights((String)tuple[col_code_idx]) == false) {
if (LOG.isTraceEnabled())
LOG.trace(String.format("Skipping AIRPORT '%s' because it does not have any flights", tuple[col_code_idx]));
continue;
}
// Update the row # so that it matches what we're actually loading
int col_id_idx = catalog_tbl.getColumnByName("AP_ID").getIndex();
tuple[col_id_idx] = (long)(row_idx + 1);
// Store Locations
int col_lat_idx = catalog_tbl.getColumnByName("AP_LATITUDE").getIndex();
int col_lon_idx = catalog_tbl.getColumnByName("AP_LONGITUDE").getIndex();
Pair<Double, Double> coords = Pair.of((Double)tuple[col_lat_idx], (Double)tuple[col_lon_idx]);
if (coords.getFirst() == null || coords.getSecond() == null) {
LOG.error(Arrays.toString(tuple));
}
assert(coords.getFirst() != null) :
String.format("Unexpected null latitude for airport '%s' [%d]", tuple[col_code_idx], col_lat_idx);
assert(coords.getSecond() != null) :
String.format("Unexpected null longitude for airport '%s' [%d]", tuple[col_code_idx], col_lon_idx);
this.airport_locations.put(tuple[col_code_idx].toString(), coords);
if (LOG.isTraceEnabled())
LOG.trace(String.format("Storing location for '%s': %s", tuple[col_code_idx], coords));
}
// Code Column -> Id Column
for (int col_code_idx : code_2_id.keySet()) {
assert(tuple[col_code_idx] != null) :
String.format("The value of the code column at '%d' is null for %s\n%s",
col_code_idx, catalog_tbl.getName(), Arrays.toString(tuple));
String code = tuple[col_code_idx].toString().trim();
if (code.length() > 0) {
Column from_column = columns.get(col_code_idx);
assert(from_column != null);
Column to_column = columns.get(code_2_id.get(col_code_idx));
assert(to_column != null) : String.format("Invalid column %s.%s", catalog_tbl.getName(), code_2_id.get(col_code_idx));
long id = (Long)tuple[code_2_id.get(col_code_idx)];
if (LOG.isTraceEnabled()) LOG.trace(String.format("Mapping %s '%s' -> %s '%d'", from_column.fullName(), code, to_column.fullName(), id));
this.profile.code_id_xref.get(to_column.getName()).put(code, id);
}
} // FOR
// Foreign Key Code -> Foreign Key Id
for (int col_code_idx : mapping_columns.keySet()) {
Column catalog_col = columns.get(col_code_idx);
assert(tuple[col_code_idx] != null || catalog_col.isNullable()) :
String.format("The code %s column at '%d' is null for %s id=%s\n%s",
catalog_col.fullName(), col_code_idx, catalog_tbl.getName(), tuple[0], Arrays.toString(tuple));
if (tuple[col_code_idx] != null) {
String code = tuple[col_code_idx].toString();
tuple[col_code_idx] = mapping_columns.get(col_code_idx).get(code);
if (LOG.isTraceEnabled())
LOG.trace(String.format("Mapped %s '%s' -> %s '%s'", catalog_col.fullName(), code, catalog_col.getForeignKey().fullName(), tuple[col_code_idx]));
}
} // FOR
for (int i = 0; i < tuple.length; i++) {
try {