perf: Add query splitting by column

[lynnagara]

If a query to Snuba involves selecting a large number of columns of raw events,
split this into 2 subqueries where:

• The first query selects only the project_id, event_id and timestamp
• The second query selects all other columns initially requested
  restricting the query to the set of project_ids, event_ids and timestamp
  range returned by the first query.

This strategy is applied instead of the exponential timesplitting in
cases where:

• There are no aggregations or groupings
• More than the minimal number of columns needed to determine the result
  set are being selected
• Ordering might not be by time

This strategy is specific to the Events DaTaSeT.

Outlined in https://www.notion.so/sentry/Eventstore-d483b27fefa9463b8723e227d86ca54a

[lynnagara]

This test is failing because an event with the same timestamp as the to_date is not returned

[lynnagara]

Need to +1 second here

[fpacifici]

@tkaemming since I believe you are making changes to api.py

[tkaemming]

Mostly small stuff, overall this looks pretty good.

[tkaemming]

There are a few conditions that are shared here and need to be True for us to even consider splitting. I think it'd be clearer if we checked those first (specifically use_split, limit and body.get('groupby') and early returned if any type of split can't/shouldn't be performed, or restructure this like:

if common conditions:
  if column conditions:
    return col_split(…)
  elif time conditions:
    return time_split(…)

return query_func(…)

It's also cheaper to check the time conditions than the column conditions (since time conditions don't require building a synthetic query) so I'd consider switching the order of evaluation.

[lynnagara]

Added the common conditions. Talked to JT about preferring col_split over time_split if both sets of conditions are met though, hence the ordering.

[tkaemming]

Oh, right — good point. Might be worth a comment in the code about why the ordering is significant since it's not immediately obvious.

[tkaemming]

Not in scope for this change, but we should absolutely not do this this way.

[tkaemming]

What's this conditional for? Should this be result['data']?

The result is always going to contain a bunch of extra stuff in addition to the result rows, so this will always be True, since parse_and_run_query (which is the only really valid de facto value of query_func) will return the value returned by raw_query:

snuba/snuba/api.py

Lines 328 to 330 in cf3e0ff

	return util.raw_query(
	validated_body, sql, clickhouse_ro, timer, stats
	)

snuba/snuba/util.py

Lines 366 to 536 in cf3e0ff

	def raw_query(body, sql, client, timer, stats=None):
	"""
	Submit a raw SQL query to clickhouse and do some post-processing on it to
	fix some of the formatting issues in the result JSON
	"""
	project_ids = to_list(body['project'])
	project_id = project_ids[0] if project_ids else 0 # TODO rate limit on every project in the list?
	stats = stats or {}
	grl, gcl, prl, pcl, use_cache, uc_max = state.get_configs([
	('global_per_second_limit', None),
	('global_concurrent_limit', 1000),
	('project_per_second_limit', 1000),
	('project_concurrent_limit', 1000),
	('use_cache', 0),
	('uncompressed_cache_max_cols', 5),
	])
	# Specific projects can have their rate limits overridden
	prl, pcl = state.get_configs([
	('project_per_second_limit_{}'.format(project_id), prl),
	('project_concurrent_limit_{}'.format(project_id), pcl),
	])
	all_confs = state.get_all_configs()
	query_settings = {
	k.split('/', 1)[1]: v
	for k, v in all_confs.items()
	if k.startswith('query_settings/')
	}
	# Experiment, if we are going to grab more than X columns worth of data,
	# don't use uncompressed_cache in clickhouse, or result cache in snuba.
	if len(all_referenced_columns(body)) > uc_max:
	query_settings['use_uncompressed_cache'] = 0
	use_cache = 0
	timer.mark('get_configs')
	query_id = md5(force_bytes(sql)).hexdigest()
	with state.deduper(query_id) as is_dupe:
	timer.mark('dedupe_wait')
	result = state.get_result(query_id) if use_cache else None
	timer.mark('cache_get')
	stats.update({
	'is_duplicate': is_dupe,
	'query_id': query_id,
	'use_cache': bool(use_cache),
	'cache_hit': bool(result)}
	),
	if result:
	status = 200
	else:
	with state.rate_limit('global', grl, gcl) as (g_allowed, g_rate, g_concurr):
	metrics.gauge('query.global_concurrent', g_concurr)
	stats.update({'global_rate': g_rate, 'global_concurrent': g_concurr})
	with state.rate_limit(project_id, prl, pcl) as (p_allowed, p_rate, p_concurr):
	stats.update({'project_rate': p_rate, 'project_concurrent': p_concurr})
	timer.mark('rate_limit')
	if g_allowed and p_allowed:
	# Experiment, reduce max threads by 1 for each extra concurrent query
	# that a project has running beyond the first one
	if 'max_threads' in query_settings and p_concurr > 1:
	maxt = query_settings['max_threads']
	query_settings['max_threads'] = max(1, maxt - p_concurr + 1)
	# Force query to use the first shard replica, which
	# should have synchronously received any cluster writes
	# before this query is run.
	consistent = body.get('consistent', False)
	stats['consistent'] = consistent
	if consistent:
	query_settings['load_balancing'] = 'in_order'
	query_settings['max_threads'] = 1
	try:
	data, meta = client.execute(
	sql,
	with_column_types=True,
	settings=query_settings,
	# All queries should already be deduplicated at this point
	# But the query_id will let us know if they aren't
	query_id=query_id
	)
	data, meta = scrub_ch_data(data, meta)
	status = 200
	if body.get('totals', False):
	assert len(data) > 0
	data, totals = data[:-1], data[-1]
	result = {'data': data, 'meta': meta, 'totals': totals}
	else:
	result = {'data': data, 'meta': meta}
	logger.debug(sql)
	timer.mark('execute')
	stats.update({
	'result_rows': len(data),
	'result_cols': len(meta),
	})
	if use_cache:
	state.set_result(query_id, result)
	timer.mark('cache_set')
	except BaseException as ex:
	error = str(ex)
	status = 500
	logger.exception("Error running query: %s\n%s", sql, error)
	if isinstance(ex, ClickHouseError):
	result = {'error': {
	'type': 'clickhouse',
	'code': ex.code,
	'message': error,
	}}
	else:
	result = {'error': {
	'type': 'unknown',
	'message': error,
	}}
	else:
	status = 429
	Reason = namedtuple('reason', 'scope name val limit')
	reasons = [
	Reason('global', 'concurrent', g_concurr, gcl),
	Reason('global', 'per-second', g_rate, grl),
	Reason('project', 'concurrent', p_concurr, pcl),
	Reason('project', 'per-second', p_rate, prl)
	]
	reason = next((r for r in reasons if r.limit is not None and r.val > r.limit), None)
	result = {'error': {
	'type': 'ratelimit',
	'message': 'rate limit exceeded',
	'detail': reason and '{r.scope} {r.name} of {r.val:.0f} exceeds limit of {r.limit:.0f}'.format(r=reason)
	}}
	stats.update(query_settings)
	if settings.RECORD_QUERIES:
	# send to redis
	state.record_query({
	'request': body,
	'sql': sql,
	'timing': timer,
	'stats': stats,
	'status': status,
	})
	# send to datadog
	tags = [
	'status:{}'.format(status),
	'referrer:{}'.format(stats.get('referrer', 'none')),
	'final:{}'.format(stats.get('final', False))
	]
	mark_tags = [
	'final:{}'.format(stats.get('final', False))
	]
	timer.send_metrics_to(metrics, tags=tags, mark_tags=mark_tags)
	result['timing'] = timer
	if settings.STATS_IN_RESPONSE or body.get('debug', False):
	result['stats'] = stats
	result['sql'] = sql
	return (result, status)

[lynnagara]

Yeah, result['data'] is what it's supposed to be. If there are no results we just go ahead and do the original query.

[tkaemming]

I think this is going to result in multiple project_id IN … in the WHERE clause because of this:

snuba/snuba/api.py

Lines 197 to 199 in cf3e0ff

	project_ids = util.to_list(body['project'])
	if project_ids:
	where_conditions.append(('project_id', 'IN', project_ids))

It probably should just replace the body['project'] value to avoid the unnecessary extra expression in the condition set.

[lynnagara]

Oops, thanks

[tkaemming]

It probably makes sense to at least comment here why this + timedelta(seconds=1) works for events. We should later consider adding something like {from,to}_date_inclusive parameters to change the behavior here, since this isn't going to work if anywhere we'd deal with higher precision timestamps.

[lynnagara]

Will add a comment. I think it's worth just changing behavior even since Sentry will also have to implement this workaround in places

[lynnagara]

I don't know why I added this, it probably doesn't matter how it's ordered we can always shrink the range we're searching

[lynnagara]

@tkaemming Think I've addressed your comments now, can you take another look?

[tkaemming]

Some minor notes inline, everything looks good functionally.

Personally, I try and minimize single-use variables (especially if they are defined far away from where they are used) because they make refactoring more challenging since you have to track down all of the places where the variable is actually referenced or mutated. If they are named just for readability reasons, I think that you can generally replace the local definition with a comment that works just as well. (Obviously this is a pretty subjective thing, you're welcome to disagree, everything works as-is.)

[tkaemming]

Not a big deal, but both total_col_count (line 23) and min_col_count are going to get eagerly evaluated since they are defined as local variables and not just part of the boolean expression that starts on line 32. (Same with remaining_offset on line 20 and orderby on line 21, these local variables may be set and never utilized.)

[tkaemming]

It seems like it would make sense to define this somewhere it can be reused as part of the expression on line 29? (Eventually, I think this will likely be something that's a method off of Dataset for any time series data — not exactly sure how that's going to look yet, though.)

[tkaemming]

The second dict shouldn't be necessary here:

Suggested change

	minimal_query = {**body, **{'selected_columns': min_cols}}
	minimal_query = {**body, 'selected_columns': min_cols}

[tkaemming]

I might consider folding the project_ids and event_ids variables into the expressions that use them to make it clearer that they won't be referenced later.

[tkaemming]

Similarly, consider folding minimal_query and second_query into the expressions they are used in.