Use bytecode generation for PageFrameReducer

[puzpuzpuz]

Is your feature request related to a problem?

Currently, we have several static methods used as PageFrameReducer for filtering and aggregation. One example:

questdb/core/src/main/java/io/questdb/griffin/engine/table/AsyncGroupByRecordCursorFactory.java

Lines 327 to 370 in 703c42d

	private static void filterAndAggregate(
	int workerId,
	@NotNull PageAddressCacheRecord record,
	@NotNull PageFrameReduceTask task,
	@NotNull SqlExecutionCircuitBreaker circuitBreaker,
	@Nullable PageFrameSequence<?> stealingFrameSequence
	) {
	final DirectLongList rows = task.getFilteredRows();
	final PageAddressCache pageAddressCache = task.getPageAddressCache();
	rows.clear();
	final long frameRowCount = task.getFrameRowCount();
	assert frameRowCount > 0;
	final AsyncGroupByAtom atom = task.getFrameSequence(AsyncGroupByAtom.class).getAtom();
	final boolean owner = stealingFrameSequence != null && stealingFrameSequence == task.getFrameSequence();
	final int slotId = atom.acquire(workerId, owner, circuitBreaker);
	final GroupByFunctionsUpdater functionUpdater = atom.getFunctionUpdater(slotId);
	final AsyncGroupByAtom.Particle particle = atom.getParticle(slotId);
	final CompiledFilter compiledFilter = atom.getCompiledFilter();
	final Function filter = atom.getFilter(slotId);
	final RecordSink mapSink = atom.getMapSink(slotId);
	try {
	if (compiledFilter == null || pageAddressCache.hasColumnTops(task.getFrameIndex())) {
	// Use Java-based filter when there is no compiled filter or in case of a page frame with column tops.
	applyFilter(filter, rows, record, frameRowCount);
	} else {
	applyCompiledFilter(compiledFilter, atom.getBindVarMemory(), atom.getBindVarFunctions(), task);
	}
	record.setRowIndex(0);
	long baseRowId = record.getRowId();
	if (!particle.isSharded()) {
	aggregateFilteredNonSharded(record, rows, baseRowId, functionUpdater, particle, mapSink);
	} else {
	aggregateFilteredSharded(record, rows, baseRowId, functionUpdater, particle, mapSink);
	}
	atom.tryShard(particle);
	} finally {
	atom.release(slotId);
	}
	}

In the case of GROUP BY, we have multiple Map implementations, as well as RecordSink, in use depending on the exact query. This leads to virtual (vtable) and interface call tables (itable) overhead for each aggregated row. We could get rid of this overhead by generating bytecode of an anonymous class similar to how it's done in RecordSinkFactory and other similar classes.

As the first step, it should be enough to generate more specific aggregation methods, e.g. aggregateFileteredSharded and aggregateFilteredNonSharded, This way, Java filter calls will still suffer from monomorphism, but at least aggregation won't.

Describe the solution you'd like.

No response

Describe alternatives you've considered.

No response

Full Name:

Andrei Pechkurov

Affiliation:

QuestDB

Additional context

No response

[puzpuzpuz]

Major part of the bottleneck was related to #4523