@Override
protected OnlineCache buildOnlineCache(Session session) {
OnlineCache onlineCache = new OnlineCache();
TransactionState txnState = txnService.getTransaction(session);
Transaction txn = txnState.getTransaction();
try {
DirectorySubspace onlineDir = smDirectory.createOrOpen(txn, ONLINE_PATH).get();
// For each online ID
for(String idStr : onlineDir.list(txn).get()) {
long onlineID = Long.parseLong(idStr);
DirectorySubspace idDir = onlineDir.open(txn, Arrays.asList(idStr)).get();
byte[] genBytes = txnState.getValue(idDir.pack(GENERATION_KEY));
long generation = Tuple2.fromBytes(genBytes).getLong(0);
// load protobuf
if(idDir.exists(txn, PROTOBUF_PATH).get()) {
DirectorySubspace protobufDir = idDir.open(txn, PROTOBUF_PATH).get();
int schemaCount = 0;
for(String schema : protobufDir.list(txn).get()) {
Long prev = onlineCache.schemaToOnline.put(schema, onlineID);
assert (prev == null) : String.format("%s, %d, %d", schema, prev, onlineID);
++schemaCount;
}
if(generation != -1) {
ProtobufReader reader = newProtobufReader();
loadProtobufChildren(txnState, protobufDir, reader, null);
loadPrimaryProtobuf(txnState, reader, onlineCache.schemaToOnline.keySet());
// Reader will have two copies of affected schemas, skip second (i.e. non-online)
AkibanInformationSchema newAIS = finishReader(reader);
validateAndFreeze(session, newAIS, generation);
buildRowDefs(session, newAIS);
onlineCache.onlineToAIS.put(onlineID, newAIS);
} else if(schemaCount != 0) {
throw new IllegalStateException("No generation but had schemas");
}
}
// Load ChangeSets
if(idDir.exists(txn, CHANGES_PATH).get()) {
DirectorySubspace changesDir = idDir.open(txn, CHANGES_PATH).get();
for(KeyValue kv : txn.getRange(Range.startsWith(changesDir.pack()))) {
ChangeSet cs = ChangeSetHelper.load(kv.getValue());
Long prev = onlineCache.tableToOnline.put(cs.getTableId(), onlineID);
assert (prev == null) : String.format("%d, %d, %d", cs.getTableId(), prev, onlineID);
onlineCache.onlineToChangeSets.put(onlineID, cs);
}