Release 1.0.6-GA

api-service/package.json

@@ -38,6 +38,7 @@
     "multiparty": "4.2.1",
     "pg": "^8.8.0",
     "prom-client": "^14.2.0",
+    "trino-client": "^0.2.2",
     "uuid": "3.1.0",
     "winston": "~2.4.3",
     "winston-daily-rotate-file": "~3.2.1"

api-service/src/configs/Config.ts

@@ -9,12 +9,21 @@ export const config = {
   "version": "1.0",
   "query_api": {
     "druid": {
+      "queryType": "realtime",
       "host": process.env.druid_host || "http://localhost",
       "port": process.env.druid_port || 8888,
       "sql_query_path": "/druid/v2/sql/",
       "native_query_path": "/druid/v2",
       "list_datasources_path": "/druid/v2/datasources",
       "submit_ingestion": "druid/indexer/v1/supervisor"
+    },
+    "lakehouse": {
+      "queryType": "datalake",
+      "host": process.env.lakehouse_host || "http://localhost",
+      "port": process.env.lakehouse_port || 8080,
+      "catalog": process.env.lakehouse_catalog || "hudi_connector",
+      "schema": process.env.lakehouse_schema || "obsrv",
+      "default_user": process.env.lakehouse_default_user || "admin"
     }
   },
   "db_connector_config": {
@@ -58,6 +67,10 @@ export const config = {
     normalDataset: "dataset",
     masterDataset: "master-dataset"
   },
+  "datasource_storage_types": {
+    druid: "druid",
+    datalake: "datalake"
+  },
   "redis_config": {
     "redis_host": process.env.redis_host || 'localhost',
     "redis_port": process.env.redis_port || 6379

api-service/src/configs/RoutesConfig.ts

@@ -170,6 +170,12 @@ export const routesConfig = {
     api_id: "api.health",
     method: "get",
     path: "/health"
+  },
+  schema_validator:{
+    api_id: "api.schema.validate",
+    method: "post",
+    path: "/data/v1/schema/validate",
+    validation_schema: "SchemaValidatorReq.json"
   }
 }
 

api-service/src/connectors/DbConnector.ts

@@ -57,7 +57,7 @@ export class DbConnector implements IConnector {
 
     public async insertRecord(table: string, fields: any) {
         await this.pool.transaction(async (dbTransaction) => {
-            await this.submit_ingestion(_.get(fields, 'ingestion_spec'), table)
+            await this.submit_ingestion(_.get(fields, 'ingestion_spec'), table, _.get(fields, "type"))
             await dbTransaction(table).insert(fields).on('query-error', (error: any) => {
                 this.log_error(OP_TYPES.INSERT, error, table, fields);
                 throw {...constants.FAILED_RECORD_CREATE, "errCode": error.code}
@@ -89,7 +89,7 @@ export class DbConnector implements IConnector {
         const existingRecord = await this.pool(table).select().where(filters).first()
         if (!_.isUndefined(existingRecord)) {
             await this.pool.transaction(async (dbTransaction) => {
-                await this.submit_ingestion(_.get(values, 'ingestion_spec'), table)
+                await this.submit_ingestion(_.get(values, 'ingestion_spec'), table, _.get(fields, "type"))
                 await dbTransaction(table).where(filters).update(schemaMerger.mergeSchema(existingRecord, values)).on('query-error', (error: any) => {
                     this.log_error(OP_TYPES.UPSERT, error, table, values);
                     throw {...constants.FAILED_RECORD_UPDATE, "errCode": error.code}
@@ -100,7 +100,7 @@ export class DbConnector implements IConnector {
             })
         } else {
             await this.pool.transaction(async (dbTransaction) => {
-                await this.submit_ingestion(_.get(values, 'ingestion_spec'), table)
+                await this.submit_ingestion(_.get(values, 'ingestion_spec'), table, _.get(fields, "type"))
                 await dbTransaction(table).insert(values).on('query-error', (error: any) => {
                     this.log_error(OP_TYPES.UPSERT, error, table, values);
                     throw {...constants.FAILED_RECORD_CREATE, "errCode": error.code}
@@ -144,8 +144,8 @@ export class DbConnector implements IConnector {
         })
     }
 
-    private async submit_ingestion(ingestion_spec: Record<string, any>, table: string) {
-        if (appConfig.table_names.datasources === table) {
+    private async submit_ingestion(ingestion_spec: Record<string, any>, table: string, storage_type: string) {
+        if (appConfig.table_names.datasources === table && storage_type === appConfig.datasource_storage_types.druid) {
             return await wrapperService.submitIngestion(ingestion_spec)
                 .catch((error: any) => {
                     console.error(constants.INGESTION_FAILED_ON_SAVE)

api-service/src/helpers/Datasources.ts

@@ -8,6 +8,7 @@ export class Datasources {
     private id: string
     private dataset_id: string
     private ingestion_spec: object
+    private type: string
     private datasource: string
     private datasource_ref: string
     private retention_period: object
@@ -30,6 +31,7 @@ export class Datasources {
         }
         this.dataset_id = payload.dataset_id
         this.ingestion_spec = payload.ingestion_spec
+        this.type = payload.type
         this.datasource = payload.datasource
         this.datasource_ref = payload.datasource_ref
         this.retention_period = payload.retentionPeriod
@@ -44,7 +46,7 @@ export class Datasources {
         this.metadata = payload.metadata
     }
     public getValues() {
-        return Object.assign(this.removeNullValues({ id: this.id, dataset_id: this.dataset_id, ingestion_spec: this.ingestion_spec, datasource: this.datasource, datasource_ref: this.datasource_ref, retention_period: this.retention_period, archival_policy: this.archival_policy, purge_policy: this.purge_policy, backup_config: this.backup_config, status: this.status, version: this.version, created_by: this.created_by, updated_by: this.updated_by, published_date: this.published_date, metadata: this.metadata }), { "updated_date": new Date })
+        return Object.assign(this.removeNullValues({ id: this.id, dataset_id: this.dataset_id, ingestion_spec: this.ingestion_spec, type: this.type, datasource: this.datasource, datasource_ref: this.datasource_ref, retention_period: this.retention_period, archival_policy: this.archival_policy, purge_policy: this.purge_policy, backup_config: this.backup_config, status: this.status, version: this.version, created_by: this.created_by, updated_by: this.updated_by, published_date: this.published_date, metadata: this.metadata }), { "updated_date": new Date })
     }
 
     public setValues() {

api-service/src/helpers/LakehouseUtil.ts

@@ -0,0 +1,51 @@
+import { Trino, BasicAuth } from 'trino-client';
+import _ from 'lodash';
+import { config } from '../configs/Config';
+
+const trino: Trino = Trino.create({
+    server: `${config.query_api.lakehouse.host}:${config.query_api.lakehouse.port}`,
+    catalog: config.query_api.lakehouse.catalog,
+    schema: config.query_api.lakehouse.schema,
+    auth: new BasicAuth(config.query_api.lakehouse.default_user),
+});
+
+
+const getFormattedData = (data: any[], columnData: any[]) => {
+    const formattedData: any[] = [];
+    for (let i = 0; i < data.length; i++) {
+        const row = data[ i ];
+        const jsonRow: any = {};
+        for (let j = 0; j < row.length; j++) {
+            // assign column only if doesn't start with _hoodie_
+            const colName = columnData[ j ];
+            if (_.startsWith(colName, "_hoodie_")) {
+                continue;
+            }
+            jsonRow[ colName ] = row[ j ];
+        }
+        formattedData.push(jsonRow);
+    }
+    return formattedData;
+}
+
+
+export const executeLakehouseQuery = async (query: string) => {   
+        const iter = await trino.query(query);
+        let queryResult: any = []
+        for await (let data of iter) {
+            if(!_.isEmpty(data.error)){
+                        throw {
+                            status: 400,
+                            message: data.error.message.replace(/line.*: /, ''),
+                            code: "BAD_REQUEST" 
+                        }
+                }
+            queryResult = [ ...queryResult, ...(data?.data?.length ?  data.data  : []) ]
+        }
+        let columns = await iter.map((r: any) => r.columns ?? []).next();
+        let finalColumns = columns.value.map((column: any) => {
+            return column.name;
+        });
+        const formattedData = getFormattedData(queryResult, finalColumns);
+        return formattedData
+}

api-service/src/helpers/ValidationService.ts

@@ -0,0 +1,12 @@
+import Ajv from "ajv";
+const validator = new Ajv({ strict: false });
+
+export const schemaValidation = (payload: Record<string, any>, schema: Record<string, any>): Record<string, any> => {
+    const isValid = validator.validate(schema, payload)
+    if (!isValid) {
+        const error: any = validator.errors;
+        const errorMessage = error[0]?.schemaPath?.replace("/", "") + " " + error[0]?.message || "Invalid Request Body";
+        return { isValid, message: errorMessage }
+    }
+    return { isValid, message: "success" }
+}

api-service/src/resources/Constants.json

@@ -102,5 +102,10 @@
     "code": "BAD_REQUEST"
   },
   "INGESTION_SUBMITTED": "ingestion spec has been submitted successfully",
-  "INGESTION_FAILED_ON_SAVE": "Failed to submit Ingestion Spec, record is not saved"
+  "INGESTION_FAILED_ON_SAVE": "Failed to submit Ingestion Spec, record is not saved",
+  "FAILED_EXECUTING_QUERY": {
+    "status": 500,
+    "message": "Something went wrong while executing the query. Please try again later.",
+    "code": "INTERNAL_SERVER_ERROR"
+}
 }

api-service/src/resources/schemas/DatasourceConfigDefault.json

@@ -17,5 +17,6 @@
     "metadata": {
         "aggregated": false,
         "granularity": "day"
-    }
+    },
+    "type": "druid"
 }