if (trace.val)
LOG.trace(String.format("[#%d-%02d] Calculating touched partitions plans for %s",
txn_id, stmt_index, catalog_stmt.fullName()));
Map<PlanFragment, PartitionSet> frag_partitions = plan.frag_partitions[stmt_index];
PartitionSet stmt_all_partitions = plan.stmt_partitions[stmt_index];
boolean has_singlepartition_plan = catalog_stmt.getHas_singlesited();
boolean is_replicated_only = this.stmt_is_replicatedonly[stmt_index];
boolean is_read_only = this.stmt_is_readonly[stmt_index];
boolean is_singlePartition = has_singlepartition_plan;
boolean is_local = true;
CatalogMap<PlanFragment> fragments = null;
// OPTIMIZATION: Fast partition look-up caching
// OPTIMIZATION: Read-only queries on replicated tables always just
// go to the local partition
// OPTIMIZATION: If we're force to be single-partitioned, pretend
// that the table is replicated
if ((this.cache_isSinglePartition != null && this.cache_isSinglePartition[stmt_index]) ||
(is_replicated_only && is_read_only) ||
(this.force_singlePartition)) {
if (trace.val) {
if (this.cache_isSinglePartition[stmt_index]) {
LOG.trace(String.format("[#%d-%02d] Using fast-lookup for %s. " +
"Skipping PartitionEstimator",
txn_id, stmt_index, catalog_stmt.fullName()));
} else {
LOG.trace(String.format("[#%d-%02d] %s is read-only and replicate-only." +
"Skipping PartitionEstimator",
txn_id, stmt_index, catalog_stmt.fullName()));
}
}
assert (has_singlepartition_plan);
if (this.cache_singlePartitionFragmentPartitions == null) {
this.cache_singlePartitionFragmentPartitions = CACHED_FRAGMENT_PARTITION_MAPS[base_partition];
}
Map<PlanFragment, PartitionSet> cached_frag_partitions = this.cache_singlePartitionFragmentPartitions.get(catalog_stmt);
if (cached_frag_partitions == null) {
cached_frag_partitions = new HashMap<PlanFragment, PartitionSet>();
PartitionSet p = this.catalogContext.getPartitionSetSingleton(base_partition);
for (PlanFragment catalog_frag : catalog_stmt.getFragments().values()) {
cached_frag_partitions.put(catalog_frag, p);
} // FOR
this.cache_singlePartitionFragmentPartitions.put(catalog_stmt, cached_frag_partitions);
}
if (plan.stmt_partitions_swap[stmt_index] == null) {
plan.stmt_partitions_swap[stmt_index] = plan.stmt_partitions[stmt_index];
plan.frag_partitions_swap[stmt_index] = plan.frag_partitions[stmt_index];
}
stmt_all_partitions = plan.stmt_partitions[stmt_index] = this.catalogContext.getPartitionSetSingleton(base_partition);
frag_partitions = plan.frag_partitions[stmt_index] = cached_frag_partitions;
}
// Otherwise figure out whether the query can execute as
// single-partitioned or not
else {
if (debug.val)
LOG.debug(String.format("[#%d-%02d] Computing touched partitions %s in txn #%d", txn_id,
stmt_index, catalog_stmt.fullName(), txn_id));
if (plan.stmt_partitions_swap[stmt_index] != null) {
stmt_all_partitions = plan.stmt_partitions[stmt_index] = plan.stmt_partitions_swap[stmt_index];
plan.stmt_partitions_swap[stmt_index] = null;
stmt_all_partitions.clear();
frag_partitions = plan.frag_partitions[stmt_index] = plan.frag_partitions_swap[stmt_index];
plan.frag_partitions_swap[stmt_index] = null;
}
try {
// OPTIMIZATION: If we were told that the transaction is suppose to be
// single-partitioned, then we will throw the single-partitioned PlanFragments
// at the PartitionEstimator to get back what partitions each PlanFragment
// will need to go to. If we get multiple partitions, then we know that we
// mispredicted and we should throw a MispredictionException
// If we originally didn't predict that it was single-partitioned, then we
// actually still need to check whether the query should be single-partitioned or not.
// This is because a query may actually just want to execute on just one
// partition (note that it could be a local partition or the remote partition).
// We'll assume that it's single-partition <<--- Can we cache that??
while (true) {
if (is_singlePartition == false) stmt_all_partitions.clear();
fragments = (is_singlePartition ? catalog_stmt.getFragments() : catalog_stmt.getMs_fragments());
if (debug.val)
LOG.debug(String.format("[#%d-%02d] Estimating for %d %s-partition fragments",
txn_id, stmt_index, fragments.size(),
(is_singlePartition ? "single" : "multi")));
// PARTITION ESTIMATOR
if (hstore_conf.site.planner_profiling && profiler != null)
ProfileMeasurementUtil.swap(profiler.plan_time, profiler.partest_time);
this.p_estimator.getAllFragmentPartitions(frag_partitions,
stmt_all_partitions,
fragments.values(),
params,
base_partition);
if (hstore_conf.site.planner_profiling && profiler != null)
ProfileMeasurementUtil.swap(profiler.partest_time, profiler.plan_time);
int stmt_all_partitions_size = stmt_all_partitions.size();
if (is_singlePartition && stmt_all_partitions_size > 1) {
// If this was suppose to be multi-partitioned, then
// we want to stop right here!!
if (predict_singlePartitioned) {
if (trace.val)
LOG.trace(String.format("Mispredicted txn #%d - Multiple Partitions %s",
txn_id, stmt_all_partitions));
mispredict = true;
break;
}
// Otherwise we can let it wrap back around and construct the fragment
// mapping for the multi-partition PlanFragments
is_singlePartition = false;
continue;
}
is_local = (stmt_all_partitions_size == 1 && stmt_all_partitions.contains(base_partition));
if (is_local == false && predict_singlePartitioned) {
// Again, this is not what was suppose to happen!
if (trace.val)
LOG.trace(String.format("Mispredicted txn #%d - Remote Partitions %s",
txn_id, stmt_all_partitions));
mispredict = true;
break;
} else if (predict_partitions.containsAll(stmt_all_partitions) == false) {
// Again, this is not what was suppose to happen!
if (trace.val)
LOG.trace(String.format("Mispredicted txn #%d - Unallocated Partitions %s / %s",
txn_id, stmt_all_partitions, predict_partitions));
mispredict = true;
break;
}
// Score! We have a plan that works!
break;
} // WHILE
// Bad Mojo!
} catch (Exception ex) {
String msg = "";
for (int i = 0; i < this.batchSize; i++) {
msg += String.format("[#%d-%02d] %s %s\n%5s\n", txn_id, i, catalog_stmt.fullName(),
catalog_stmt.getSqltext(), Arrays.toString(batchArgs[i].toArray()));
} // FOR
LOG.fatal("\n" + msg);
throw new RuntimeException("Unexpected error when planning " + catalog_stmt.fullName(), ex);
}
}
if (debug.val)
LOG.debug(String.format("[#%d-%02d] is_singlepartition=%s, partitions=%s",
txn_id, stmt_index, is_singlePartition, stmt_all_partitions));
// Get a sorted list of the PlanFragments that we need to execute
// for this query
if (is_singlePartition) {
if (this.sorted_singlep_fragments[stmt_index] == null) {
this.sorted_singlep_fragments[stmt_index] = PlanNodeUtil.getSortedPlanFragments(catalog_stmt, true);
}
plan.frag_list[stmt_index] = this.sorted_singlep_fragments[stmt_index];
// Only mark that we touched these partitions if the Statement
// is not on a replicated table or it's not read-only
if (is_replicated_only == false || is_read_only == false) {
touched_partitions.put(stmt_all_partitions.get());
}
}
// Distributed Query
else {
if (this.sorted_multip_fragments[stmt_index] == null) {
this.sorted_multip_fragments[stmt_index] = PlanNodeUtil.getSortedPlanFragments(catalog_stmt, false);
}
plan.frag_list[stmt_index] = this.sorted_multip_fragments[stmt_index];
// Always mark that we are touching these partitions
touched_partitions.put(stmt_all_partitions.values());
// Note that will want to update is_singlePartitioned here for non-readonly replicated
// querys when we have a one partition cluster because those queries don't have
// single-partition query plans
// if (this.num_partitions == 1 && is_replicated_only && is_read_only == false) {
// is_singlePartition = true;
// }
}
plan.readonly = plan.readonly && catalog_stmt.getReadonly();
plan.all_singlepartitioned = plan.all_singlepartitioned && is_singlePartition;
plan.all_local = plan.all_local && is_local;
// Keep track of whether the current query in the batch was
// single-partitioned or not
plan.singlepartition_bitmap[stmt_index] = is_singlePartition;
// Misprediction!!
if (mispredict) {
// If this is the first Statement in the batch that hits the mispredict,
// then we need to create the histogram and populate it with the
// partitions from the previous queries
int start_idx = stmt_index;
if (mispredict_h == null) {
mispredict_h = new FastIntHistogram();
start_idx = 0;
}
for (int i = start_idx; i <= stmt_index; i++) {
if (debug.val)
LOG.debug(String.format("Pending mispredict for txn #%d. " +
"Checking whether to add partitions for batch statement %02d",
txn_id, i));
// Make sure that we don't count the local partition if it
// was reading a replicated table.
if (this.stmt_is_replicatedonly[i] == false ||
(this.stmt_is_replicatedonly[i] && this.stmt_is_readonly[i] == false)) {
if (trace.val)
LOG.trace(String.format("%s touches non-replicated table. " +
"Including %d partitions in mispredict histogram for txn #%d",
this.catalog_stmts[i].fullName(), plan.stmt_partitions[i].size(), txn_id));
mispredict_h.put(plan.stmt_partitions[i]);
}
} // FOR
continue;
}
// ----------------------
// DEBUG DUMP
// ----------------------
if (debug.val) {
List<PlanFragment> _fragments = null;
if (is_singlePartition && this.sorted_singlep_fragments[stmt_index] != null) {
_fragments = this.sorted_singlep_fragments[stmt_index];
} else {
_fragments = this.sorted_multip_fragments[stmt_index];
}
Map<?, ?> maps[] = new Map[_fragments.size() + 1];
int ii = 0;
for (PlanFragment catalog_frag : _fragments) {
Map<String, Object> m = new LinkedHashMap<String, Object>();
PartitionSet p = plan.frag_partitions[stmt_index].get(catalog_frag);
boolean frag_local = (p.size() == 1 && p.contains(base_partition));
m.put(String.format("[%02d] Fragment", ii), catalog_frag.fullName());
m.put(String.format(" Partitions"), p);
m.put(String.format(" IsLocal"), frag_local);
ii++;
maps[ii] = m;