Feature(wren-ui) duckdb onboarding process (#38)

* implement deploy mdl

* feat(wren-ui): onboarding process support duckdb as datasource

* fix lint
This commit is contained in:
Andy Yen
2024-03-27 11:54:38 +08:00
committed by GitHub
parent e73cf7acbc
commit 26d4da41e3
16 changed files with 503 additions and 65 deletions
@@ -13,11 +13,34 @@ exports.up = function (knex) {
table.string('display_name').comment('project display name');
// bq
table.string('project_id').comment('gcp project id, big query specific');
table
.string('project_id')
.nullable()
.comment('gcp project id, big query specific');
table
.text('credentials')
.nullable()
.comment('project credentials, big query specific');
table.string('dataset_id').comment('big query datasetId');
table.string('dataset_id').nullable().comment('big query datasetId');
// duckdb
table
.jsonb('init_sql')
.nullable()
.comment('init sql for establishing duckdb environment');
// knex jsonb ref: https://knexjs.org/guide/schema-builder.html#json
table
.jsonb('extensions')
.nullable()
.comment(
'duckdb extensions, will be a array-like string like, eg: ["extension1", "extension2"]',
);
table
.jsonb('configurations')
.nullable()
.comment(
'duckdb configurations that can be set in session, eg: { "key1": "value1", "key2": "value2" }',
);
// not sure to store or not, the catalog & schema in the manifest
table.string('catalog').comment('catalog name');
@@ -1,4 +1,4 @@
import axios, { AxiosError } from 'axios';
import axios from 'axios';
import { Manifest } from '../mdl/type';
import { getLogger } from '@server/utils';
@@ -1,4 +1,4 @@
import axios, { AxiosError } from 'axios';
import axios from 'axios';
import { Manifest } from '../mdl/type';
import { getLogger } from '@server/utils';
@@ -20,11 +20,6 @@ export interface DeployResponse {
error?: string;
}
enum WrenEngineSystemStatus {
READY = 'READY',
PREPARE = 'PREPARE',
}
interface DeployPayload {
manifest: Manifest;
version: string;
@@ -35,12 +30,27 @@ export interface deployData {
hash: string;
}
export interface ColumnMetadata {
name: string;
type: string;
}
export interface QueryResponse {
columns: ColumnMetadata[];
data: any[][];
}
export interface IWrenEngineAdaptor {
deploy(deployData: deployData): Promise<DeployResponse>;
initDatabase(sql: string): Promise<void>;
putSessionProps(props: Record<string, any>): Promise<void>;
queryDuckdb(sql: string): Promise<QueryResponse>;
}
export class WrenEngineAdaptor implements IWrenEngineAdaptor {
private readonly wrenEngineBaseEndpoint: string;
private sessionPropsUrlPath = '/v1/data-source/DuckDB/settings/session-sql';
private queryDuckdbUrlPath = '/v1/data-source/DuckDB/query';
private initSqlUrlPath = '/v1/data-source/DuckDB/settings/init-sql';
constructor({ wrenEngineEndpoint }: { wrenEngineEndpoint: string }) {
this.wrenEngineBaseEndpoint = wrenEngineEndpoint;
}
@@ -71,6 +81,56 @@ export class WrenEngineAdaptor implements IWrenEngineAdaptor {
}
}
public async initDatabase(sql) {
try {
const url = new URL(this.initSqlUrlPath, this.wrenEngineBaseEndpoint);
logger.debug(`Endpoint: ${url.href}`);
const headers = {
'Content-Type': 'text/plain; charset=utf-8',
};
await axios.put(url.href, sql, { headers });
} catch (err: any) {
logger.debug(`Got error when init database: ${err.message}`);
throw err;
}
}
public async putSessionProps(props: Record<string, any>) {
const setSessionStatements = Object.entries(props)
.map(([key, value]) => {
return `SET ${key} = '${value}';`;
})
.join('\n');
try {
const url = new URL(
this.sessionPropsUrlPath,
this.wrenEngineBaseEndpoint,
);
logger.debug(`Endpoint: ${url.href}`);
const headers = {
'Content-Type': 'text/plain; charset=utf-8',
};
await axios.put(url.href, setSessionStatements, { headers });
} catch (err: any) {
logger.debug(`Got error when put session props: ${err.message}`);
throw err;
}
}
public async queryDuckdb(sql: string): Promise<QueryResponse> {
try {
const url = new URL(this.queryDuckdbUrlPath, this.wrenEngineBaseEndpoint);
const headers = {
'Content-Type': 'text/plain; charset=utf-8',
};
const res = await axios.post(url.href, sql, { headers });
return res.data as QueryResponse;
} catch (err: any) {
logger.debug(`Got error when querying duckdb: ${err.message}`);
throw err;
}
}
private async getDeployStatus(): Promise<WrenEngineDeployStatusResponse> {
try {
const res = await axios.get(
@@ -53,6 +53,10 @@ export class BQConnector
this.bq = new BigQuery(bqOptions);
}
public async prepare() {
return;
}
public async connect() {
try {
await this.bq.query('SELECT 1;');
@@ -1,6 +1,7 @@
import { CompactTable } from '../types';
export interface IConnector<T, C> {
prepare(prepareOptions: any): Promise<void>;
connect(): Promise<boolean>;
listTables(listTableOptions: any): Promise<CompactTable[] | T[]>;
listConstraints(listConstraintOptions: any): Promise<[] | C[]>;
@@ -0,0 +1,112 @@
import {
IWrenEngineAdaptor,
QueryResponse,
} from '../adaptors/wrenEngineAdaptor';
import { CompactTable } from '../types';
import { IConnector } from './connector';
import { getLogger } from '@server/utils';
const logger = getLogger('DuckDBConnector');
logger.level = 'debug';
export interface DuckDBPrepareOptions {
initSql: string;
sessionProps: Record<string, any>;
}
export interface DuckDBListTableOptions {
format?: boolean;
}
export interface DuckDBColumnResponse {
table_catalog: string;
table_schema: string;
table_name: string;
column_name: string;
ordinal_position: string;
is_nullable: string;
data_type: string;
}
export class DuckDBConnector
implements IConnector<DuckDBColumnResponse[], any[]>
{
private wrenEngineAdaptor: IWrenEngineAdaptor;
constructor({
wrenEngineAdaptor,
}: {
wrenEngineAdaptor: IWrenEngineAdaptor;
}) {
this.wrenEngineAdaptor = wrenEngineAdaptor;
}
public async prepare(prepareOptions: DuckDBPrepareOptions): Promise<void> {
const { initSql, sessionProps } = prepareOptions;
await this.wrenEngineAdaptor.initDatabase(initSql);
await this.wrenEngineAdaptor.putSessionProps(sessionProps);
}
public async connect(): Promise<boolean> {
const sql = 'SELECT 1;';
try {
await this.wrenEngineAdaptor.queryDuckdb(sql);
return true;
} catch (_err) {
return false;
}
}
public async listTables(listTableOptions: DuckDBListTableOptions) {
const sql =
'SELECT \
table_catalog, table_schema, table_name, column_name, ordinal_position, is_nullable, data_type\
FROM INFORMATION_SCHEMA.COLUMNS;';
const response = await this.wrenEngineAdaptor.queryDuckdb(sql);
if (listTableOptions.format) {
return this.formatToCompactTable(response);
}
return response.data;
}
public async listConstraints(): Promise<any[]> {
return [];
}
private formatToCompactTable(columns: QueryResponse): CompactTable[] {
return columns.data.reduce((acc: CompactTable[], row: any) => {
const [
table_catalog,
table_schema,
table_name,
column_name,
ordinal_position,
is_nullable,
data_type,
] = row;
let table = acc.find(
(t) => t.name === table_name && t.properties.schema === table_schema,
);
if (!table) {
table = {
name: table_name,
description: '',
columns: [],
properties: {
schema: table_schema,
catalog: table_catalog,
},
};
acc.push(table);
}
table.columns.push({
name: column_name,
type: data_type,
notNull: is_nullable.toLocaleLowerCase() !== 'yes',
description: '',
properties: {
ordinalPosition: ordinal_position,
},
});
return acc;
}, []);
}
}
@@ -116,7 +116,7 @@ export class BaseRepository<T> implements IBasicRepository<T> {
const result = await executer(this.tableName)
.insert(data.map(this.transformToDBData))
.returning('*');
return result;
return result.map((data) => this.transformFromDBData(data));
}
public async updateOne(
@@ -1,13 +1,28 @@
import { Knex } from 'knex';
import { BaseRepository, IBasicRepository } from './baseRepository';
import {
camelCase,
isPlainObject,
mapKeys,
mapValues,
snakeCase,
} from 'lodash';
export interface Project {
id: number; // ID
type: string; // Project datasource type. ex: bigquery, mysql, postgresql, mongodb, etc
displayName: string; // Project display name
// bq
projectId: string; // BigQuery project id
datasetId: string; // BigQuery datasetId
credentials: string; // Project credentials, big query specific
// duckdb
initSql: string; // DuckDB init sql
configurations: Record<string, any>; // DuckDB configurations
extensions: string[];
catalog: string; // Catalog name
schema: string; // Schema name
sampleDataset: string; // Sample dataset name
@@ -35,4 +50,34 @@ export class ProjectRepository
}
return projects[0];
}
public override transformFromDBData: (data: any) => Project = (data: any) => {
if (!isPlainObject(data)) {
throw new Error('Unexpected dbdata');
}
const camelCaseData = mapKeys(data, (_value, key) => camelCase(key));
const formattedData = mapValues(camelCaseData, (value, key) => {
if (['configurations', 'extensions'].includes(key)) {
return JSON.parse(value);
}
return value;
});
return formattedData as Project;
};
public override transformToDBData: (data: Project) => any = (
data: Project,
) => {
if (!isPlainObject(data)) {
throw new Error('Unexpected dbdata');
}
const snakeCaseData = mapKeys(data, (_value, key) => snakeCase(key));
const formattedData = mapValues(snakeCaseData, (value, key) => {
if (['configurations', 'extensions'].includes(key)) {
return JSON.stringify(value);
}
return value;
});
return formattedData;
};
}
@@ -12,9 +12,16 @@ import {
RelationData,
AnalysisRelationInfo,
RelationType,
CompactTable,
} from '../types';
import { getLogger, Encryptor } from '@server/utils';
import { Model, ModelColumn, Project, Relation } from '../repositories';
import {
DuckDBConnector,
DuckDBListTableOptions,
DuckDBPrepareOptions,
} from '../connectors/duckdbConnector';
import { IConnector } from '../connectors/connector';
const logger = getLogger('DataSourceResolver');
logger.level = 'debug';
@@ -73,19 +80,34 @@ export class ProjectResolver {
const { type, properties } = args.data;
if (type === DataSourceName.BIG_QUERY) {
await this.saveBigQueryDataSource(properties, ctx);
return args.data;
} else if (type === DataSourceName.DUCKDB) {
await this.saveDuckDBDataSource(properties, ctx);
}
return args.data;
}
public async listDataSourceTables(_root: any, _arg, ctx: IContext) {
const project = await ctx.projectService.getCurrentProject();
const filePath = await ctx.projectService.getCredentialFilePath(project);
const connector = await this.getBQConnector(project, filePath);
const listTableOptions: BQListTableOptions = {
datasetId: project.datasetId,
format: true,
};
return await connector.listTables(listTableOptions);
let connector: IConnector<any, any>;
let listTableOptions: any;
if (project.type === DataSourceName.BIG_QUERY) {
const filePath = await ctx.projectService.getCredentialFilePath(project);
connector = await this.getBQConnector(project, filePath);
listTableOptions = {
datasetId: project.datasetId,
format: true,
} as BQListTableOptions;
} else {
connector = new DuckDBConnector({
wrenEngineAdaptor: ctx.wrenEngineAdaptor,
});
listTableOptions = { format: true } as DuckDBListTableOptions;
}
const tables = (await connector.listTables(
listTableOptions,
)) as CompactTable[];
logger.debug(tables);
return tables;
}
public async saveTables(
@@ -99,52 +121,86 @@ export class ProjectResolver {
// get current project
const project = await ctx.projectService.getCurrentProject();
const filePath = await ctx.projectService.getCredentialFilePath(project);
const projectId = project.id;
// get columns with descriptions
const connector = await this.getBQConnector(project, filePath);
const listTableOptions: BQListTableOptions = {
datasetId: project.datasetId,
format: false,
};
const dataSourceColumns = await connector.listTables(listTableOptions);
let connector: IConnector<any, any>;
let listTableOptions: any;
let dataSourceColumns: any;
if (project.type === DataSourceName.BIG_QUERY) {
const filePath = await ctx.projectService.getCredentialFilePath(project);
connector = await this.getBQConnector(project, filePath);
listTableOptions = {
datasetId: project.datasetId,
format: false,
} as BQListTableOptions;
dataSourceColumns = (await connector.listTables(
listTableOptions,
)) as BQColumnResponse[];
} else {
connector = new DuckDBConnector({
wrenEngineAdaptor: ctx.wrenEngineAdaptor,
});
listTableOptions = { format: true } as DuckDBListTableOptions;
dataSourceColumns = (await connector.listTables(
listTableOptions,
)) as CompactTable[];
}
// delete existing models and columns
await this.resetCurrentProjectModel(ctx, projectId);
// create models
const tableDescriptions = (dataSourceColumns as BQColumnResponse[])
.filter((col) => col.table_description)
.reduce((acc, column) => {
acc[column.table_name] = column.table_description;
return acc;
}, {});
const models = await this.createModels(
tables,
projectId,
ctx,
tableDescriptions,
);
// create columns
const columns = await this.createAllColumnsInDataSource(
tables,
models,
dataSourceColumns as BQColumnResponse[],
ctx,
);
let models: Model[];
let columns: ModelColumn[];
if (project.type === DataSourceName.BIG_QUERY) {
models = await this.createBigQueryModels(
tables,
projectId,
ctx,
dataSourceColumns,
);
// create columns
columns = await this.createBigQueryColumns(
tables,
models,
dataSourceColumns as BQColumnResponse[],
ctx,
);
} else {
models = await this.createDuckDBModels(
tables,
projectId,
ctx,
dataSourceColumns as CompactTable[],
);
logger.debug({ models });
// create columns
columns = await this.createDuckDBColumns(
tables,
models,
dataSourceColumns as CompactTable[],
ctx,
);
}
this.deploy(ctx);
return { models: models, columns };
}
public async autoGenerateRelation(_root: any, _arg: any, ctx: IContext) {
const project = await ctx.projectService.getCurrentProject();
const filePath = await ctx.projectService.getCredentialFilePath(project);
const models = await ctx.modelRepository.findAllBy({
projectId: project.id,
});
// Duckdb: skip auto generate relation
if (project.type === DataSourceName.DUCKDB) {
return models.map(({ id, sourceTableName }) => ({
id,
name: sourceTableName,
relations: [],
}));
}
const filePath = await ctx.projectService.getCredentialFilePath(project);
const connector = await this.getBQConnector(project, filePath);
const listConstraintOptions = {
datasetId: project.datasetId,
@@ -279,7 +335,7 @@ export class ProjectResolver {
return new BQConnector(connectionOption);
}
private async createAllColumnsInDataSource(
private async createBigQueryColumns(
tables: string[],
models: Model[],
dataSourceColumns: BQColumnResponse[],
@@ -320,14 +376,58 @@ export class ProjectResolver {
return columns;
}
private async createModels(
private async createDuckDBColumns(
tables: string[],
models: Model[],
compactTables: CompactTable[],
ctx: IContext,
) {
const columnValues = tables.reduce((acc, tableName) => {
const modelId = models.find((m) => m.sourceTableName === tableName)?.id;
if (!modelId) {
throw new Error(`Model not found: ${tableName}`);
}
const compactColumns = compactTables.find(
(table) => table.name === tableName,
)?.columns;
if (!compactColumns) {
throw new Error('Table not found');
}
for (const compactColumn of compactColumns) {
const columnName = compactColumn.name;
const columnValue = {
modelId,
isCalculated: false,
displayName: columnName,
sourceColumnName: columnName,
referenceName: columnName,
type: compactColumn.type || 'string',
notNull: compactColumn.notNull,
isPk: false,
properties: JSON.stringify(compactColumn.properties),
} as Partial<ModelColumn>;
acc.push(columnValue);
}
return acc;
}, []);
const columns = await ctx.modelColumnRepository.createMany(columnValues);
return columns;
}
private async createBigQueryModels(
tables: string[],
id: number,
ctx: IContext,
tableDescriptions: { [key: string]: string },
dataSourceColumns: BQColumnResponse[],
) {
const tableDescriptionMap = dataSourceColumns
.filter((col) => col.table_description)
.reduce((acc, column) => {
acc[column.table_name] = column.table_description;
return acc;
}, {});
const modelValues = tables.map((tableName) => {
const description = tableDescriptions[tableName];
const description = tableDescriptionMap[tableName];
const model = {
projectId: id,
displayName: tableName, //use table name as displayName, referenceName and tableName
@@ -341,14 +441,82 @@ export class ProjectResolver {
return model;
});
const models = await Promise.all(
modelValues.map(
async (modelValue) => await ctx.modelRepository.createOne(modelValue),
),
);
const models = await ctx.modelRepository.createMany(modelValues);
return models;
}
private async createDuckDBModels(
tables: string[],
id: number,
ctx: IContext,
compactTables: CompactTable[],
) {
const modelValues = tables.map((tableName) => {
const compactTable = compactTables.find(
(table) => table.name === tableName,
);
const properties = compactTable.properties
? JSON.stringify(compactTable.properties)
: null;
const model = {
projectId: id,
displayName: tableName, //use table name as displayName, referenceName and tableName
referenceName: tableName,
sourceTableName: tableName,
refSql: `select * from ${compactTable.properties.schema}.${tableName}`,
cached: false,
refreshTime: null,
properties,
} as Partial<Model>;
return model;
});
const models = await ctx.modelRepository.createMany(modelValues);
return models;
}
private async saveDuckDBDataSource(properties: any, ctx: IContext) {
const { displayName, extensions, configurations } = properties;
const initSql = this.concatInitSql(properties.initSql, extensions);
const connector = new DuckDBConnector({
wrenEngineAdaptor: ctx.wrenEngineAdaptor,
});
// prepare duckdb environment in wren-engine
const prepareOption = {
sessionProps: configurations,
initSql,
} as DuckDBPrepareOptions;
await connector.prepare(prepareOption);
// check DataSource is valid and can connect to it
const connected = await connector.connect();
if (!connected) {
throw new Error('Can not connect to data source');
}
// check can list dataset table
try {
await connector.listTables({ format: false });
} catch (_e) {
throw new Error('Can not list tables in dataset');
}
// remove existing datasource
await this.removeCurrentProject(ctx);
// save DataSource to database
const project = await ctx.projectRepository.createOne({
displayName,
schema: 'tbd',
catalog: 'tbd',
type: DataSourceName.DUCKDB,
initSql,
configurations,
extensions,
});
return project;
}
private async saveBigQueryDataSource(properties: any, ctx: IContext) {
const { displayName, projectId, datasetId, credentials } = properties;
const { config } = ctx;
@@ -363,6 +531,7 @@ export class ProjectResolver {
keyFilename: filePath,
};
const connector = new BQConnector(connectionOption);
await connector.prepare();
const connected = await connector.connect();
if (!connected) {
throw new Error('Can not connect to data source');
@@ -378,13 +547,7 @@ export class ProjectResolver {
const encryptedCredentials = encryptor.encrypt(credentials);
// remove existing datasource
try {
const currentProject = await ctx.projectRepository.getCurrentProject();
await this.resetCurrentProjectModel(ctx, currentProject.id);
await ctx.projectRepository.deleteOne(currentProject.id);
} catch (_err: any) {
// do nothing
}
await this.removeCurrentProject(ctx);
// TODO: add displayName, schema, catalog to the DataSource, depends on the MDL structure
const project = await ctx.projectRepository.createOne({
@@ -425,4 +588,23 @@ export class ProjectResolver {
await ctx.modelColumnRepository.deleteByModelIds(modelIds);
await ctx.modelRepository.deleteMany(modelIds);
}
private concatInitSql(initSql: string, extensions: string[]) {
const installExtensions = extensions
.map((ext) => `INSTALL ${ext};`)
.join('\n');
return `${installExtensions}\n${initSql}`;
}
private async removeCurrentProject(ctx) {
let currentProject: Project;
try {
currentProject = await ctx.projectRepository.getCurrentProject();
} catch (_err: any) {
// no project found
return;
}
await this.resetCurrentProjectModel(ctx, currentProject.id);
await ctx.projectRepository.deleteOne(currentProject.id);
}
}
+3
View File
@@ -5,6 +5,7 @@ export const typeDefs = gql`
enum DataSourceName {
BIG_QUERY
DUCKDB
}
type UsableDataSource {
@@ -25,6 +26,7 @@ export const typeDefs = gql`
type CompactTable {
name: String!
columns: [CompactColumn!]!
properties: JSON
}
input MDLModelSubmitInput {
@@ -84,6 +86,7 @@ export const typeDefs = gql`
type CompactColumn {
name: String!
type: String!
properties: JSON
}
input CustomFieldInput {
@@ -1,3 +1,4 @@
import { IWrenEngineAdaptor } from '../adaptors/wrenEngineAdaptor';
import { IConfig } from '../config';
import {
IModelColumnRepository,
@@ -14,6 +15,9 @@ import { IProjectService } from '../services/projectService';
export interface IContext {
config: IConfig;
// adaptor
wrenEngineAdaptor: IWrenEngineAdaptor;
// services
projectService: IProjectService;
modelService: IModelService;
@@ -1,5 +1,6 @@
export enum DataSourceName {
BIG_QUERY = 'BIG_QUERY',
DUCKDB = 'DUCKDB',
}
interface BaseDataSource {
+2
View File
@@ -38,10 +38,12 @@ export interface CompactColumn {
type: string;
notNull: boolean;
description?: string;
properties?: Record<string, any>;
}
export interface CompactTable {
name: string;
columns: CompactColumn[];
description?: string;
properties?: Record<string, any>;
}
@@ -1,4 +1,3 @@
import { IConfig } from '../../config';
import { Encryptor } from '../encryptor';
import crypto from 'crypto';
@@ -1,7 +1,6 @@
import { ReactFlowInstance, useReactFlow } from 'reactflow';
import styled from 'styled-components';
import MarkerHandle from './MarkerHandle';
import CustomPopover from '../CustomPopover';
const NodeColumn = styled.div`
position: relative;
+3
View File
@@ -77,6 +77,9 @@ const apolloServer: ApolloServer = new ApolloServer({
context: (): IContext => ({
config: serverConfig,
// adaptor
wrenEngineAdaptor,
// services
projectService,
modelService,