Closed eduohe closed 1 month ago
Created this new SCD TYPE 1 table in Databricks:
CREATE OR REFRESH STREAMING LIVE TABLE silver_action_function_call_methods
COMMENT "Stream of action receipt function call methods. This table is a SCD (slow changing dimension) type 1 that upsert rows to have the latest record"
TBLPROPERTIES ("quality" = "silver", delta.autoOptimize.optimizeWrite = true, delta.autoOptimize.autoCompact = true);
APPLY CHANGES INTO live.silver_action_function_call_methods
FROM (
SELECT
block_date,
block_height,
block_timestamp,
block_timestamp_utc,
block_hash,
chunk_hash,
shard_id,
receipt_id,
receipt_receiver_account_id,
args:method_name as method_name,
try_cast(unbase64(args:args_base64) AS STRING) AS args
FROM STREAM(live.silver_action_receipt_actions) ara
WHERE ara.action_kind = 'FUNCTION_CALL'
)
KEYS (receipt_receiver_account_id, method_name)
IGNORE NULL UPDATES
SEQUENCE BY block_timestamp
STORED AS SCD TYPE 1
and this Cloud Function:
const functions = require('@google-cloud/functions-framework');
const genson = require('genson-js');
functions.http('get_schemas', (req, res) => {
res.set('Access-Control-Allow-Origin', '*');
if (req.method === 'OPTIONS') {
// Send response to OPTIONS requests
res.set('Access-Control-Allow-Methods', 'GET');
res.set('Access-Control-Allow-Headers', 'Content-Type');
res.set('Access-Control-Max-Age', '3600');
res.status(204).send('');
return;
}
const filter = req.body.filter || req.query.filter;
const validPattern = /^[A-Za-z0-9%.\-_*]+$/;
if (!filter.match(validPattern)) {
return res.status(400).json({ err: "Only letters, numbers, %, ., -, _, and * are allowed!" });
}
const { DBSQLClient } = require('@databricks/sql');
var token = process.env.DATABRICKS_ACCESS_TOKEN;
var server_hostname = "4221960800361869.9.gcp.databricks.com";
var http_path = "/sql/1.0/warehouses/71275359352433a0";
const client = new DBSQLClient();
client.connect(
options = {
token: token,
host: server_hostname,
path: http_path
}).then(
async client => {
const session = await client.openSession();
const queryOperation = await session.executeStatement(
statement = `SELECT method_name, max(args) as args FROM mainnet.silver_action_function_call_methods ara WHERE receipt_receiver_account_id LIKE '${filter.replace('*', '%')}' AND status <> 'FAILURE' AND method_name IS NOT NULL GROUP BY ALL ORDER BY method_name;`,
options = { runAsync: true });
const result = await queryOperation.fetchAll();
await queryOperation.close();
let resPayload = []
result.forEach((row) => {
let obj = {}
try {
obj = JSON.parse(row.args);
} catch(e) {
// Ignore no json schema
}
const schema = genson.createSchema(obj);
resPayload.push({
method_name: row.method_name,
schema: schema
});
});
await session.close();
await client.close();
res.status(200).send(resPayload);
}).catch(error => {
console.log(error);
res.status(500).send(error);
});
});
Sample usage:
curl --location 'https://europe-west1-pagoda-data-stack-prod.cloudfunctions.net/queryapi_wizard' \
--header 'Content-Type: application/json' \
--data '{
"filter": "*pool.near"
}'
Create a Cloud Function endpoint that that returns method args schemas for receipt_receiver_account_id allowing filters like
*pool.near
Note: Latency should be low enough for use cases when a developer deploys a smart contract and wants to create an indexer in QueryAPI