async function buildMultipartParquet(
	columns,
	data,
	tmpDir,
	outDir,
	outputFilename,
	batchSize = 1000000
) {
	// log.debug(`Building parquet file ${outputFilename}`);
	// let { meta: fn_meta, done: fn_done } = log.measure('buildMultipartParquet');
	// fn_meta('output filename', outputFilename);

	let batchNum = 0;
	const outputSubpath = outputFilename.split('.parquet')[0];
	let tmpFilenames = [];
	let rowCount = 0;

	const flush = async (results) => {
		// log.debug(`Flushing batch ${batchNum} with ${results.length} rows`);
		// let { meta, done } = log.measure('flush');
		// meta('batch number', batchNum);

		// Convert JS Objects -> Arrow
		const vectorized = Object.fromEntries(
			columns.map((c) => [
				c.name,
				convertArrayToVector(
					c,
					results.map((i) => i[c.name] ?? null)
				)
			])
		);
		const table = tableFromArrays(vectorized);
		for (const field of table.schema.fields) {
			field.nullable = true;
		}

		// Converts the arrow table to a buffer
		const IPC = tableToIPC(table, 'stream');

		const writerProperties = new WriterPropertiesBuilder().setCompression(Compression.ZSTD).build();
		// Converts the arrow buffer to a parquet buffer
		// This can be slow; it includes the compression step

		const parquetBuffer = writeParquet(Table.fromIPCStream(IPC), writerProperties);

		const tempFilename = path.join(tmpDir, `${outputSubpath}.${batchNum}.parquet`);
		await fs.mkdir(path.dirname(tempFilename), { recursive: true });
		await fs.writeFile(tempFilename, parquetBuffer);

		tmpFilenames.push(tempFilename);
		rowCount += results.length;

		// done();
		// log.debug(`Flushed batch ${batchNum} with ${results.length} rows`);

		batchNum++;
	};

	if (typeof data === 'function') data = data();
	if (data instanceof Promise) data = await data;
	// Data is an array, but not a nested one
	// We expect a set of result sets;
	if (Array.isArray(data) && !Array.isArray(data[0])) data = [data];
	if (Array.isArray(data)) {
		const arrays = data;
		data = (function* () {
			for (const results of arrays) {
				for (const batch of chunk(results, batchSize)) {
					yield batch;
				}
			}
		})();
	}

	// log.debug('Reading rows from a generator object');
	// let { meta, done } = log.measure('buildMultipartParquet');
	// meta('batch number', batchNum);

	const currentBatch = [];
	for await (const results of data) {
		for (const result of results) currentBatch.push(result);

		if (currentBatch.length >= batchSize) {
			// done();
			// log.debug(`Flushing batch ${batchNum} with ${currentBatch.length} rows`);
			await flush(currentBatch);
			currentBatch.length = 0;
			// ({ meta, done } = log.measure('buildMultipartParquet'));
			// meta('batch number', batchNum);
		}
	}

	// done();
	// log.debug(`Flushing batch ${batchNum} with ${currentBatch.length} rows`);

	if (currentBatch.length) await flush(currentBatch);

	if (!tmpFilenames.length) return 0;

	await initDB();

	const outputFilepath = path.join(outDir, outputFilename);

	const parquetFiles = tmpFilenames.map((filename) => `'${filename.replaceAll('\\', '/')}'`);

	const select = `SELECT * FROM read_parquet([${parquetFiles.join(',')}])`;
	const copy = `COPY (${select}) TO '${outputFilepath}' (FORMAT 'PARQUET', CODEC 'ZSTD', USE_TMP_FILE false);`;

	await fs.mkdir(path.dirname(outputFilepath), { recursive: true });
	await query(copy);

	await fs.chmod(outputFilepath, 0o644);

	const { size } = await fs.stat(outputFilepath);
	if (size > 100 * 1024 * 1024) {
		console.warn(
			chalk.yellow(` Estimated disk size is ${Intl.NumberFormat().format(size / (1024 * 1024))}mb.`)
		);
	}

	const score =
		rowCount *
		columnsToScore(
			columns.map(({ name, evidenceType }) => ({
				name,
				type:
					evidenceType === 'number'
						? 'DOUBLE'
						: evidenceType === 'boolean'
							? 'BOOLEAN'
							: evidenceType === 'date'
								? 'TIMESTAMP'
								: 'VARCHAR'
			}))
		);

	if (score > 100 * 1024 * 1024) {
		console.warn(
			chalk.yellow(
				` WARNING: Estimated output size is ${Intl.NumberFormat().format(
					score / (1024 * 1024)
				)}mb uncompressed. This may cause client-side performance issues.`
			)
		);
	}

	for (const tmpFile of tmpFilenames) {
		await fs.rm(tmpFile, { force: true });
	}
	await emptyDbFs('*');

	// fn_done();

	return rowCount;
}
@template T
@param {{name: string, evidenceType: string}[]} columns
@param {Generator<T[] | Promise<T[]> | T[] | Promise<T[]>} data
@param {string} tmpDir
@param {string} outDir
@param {string} outputFilename
@param {number} [batchSize]
@returns {Promise<number>} Number of rows