Merge pull request #15 from dataform-co/remove-sql-package

Remove sql package usage

dataform.json

@@ -3,4 +3,4 @@
     "assertionSchema": "segment_dataform_package",
     "warehouse": "bigquery",
     "gcloudProjectId": "dataform-corp"
-}
+}

includes/sessionized_events.js

@@ -1,4 +1,4 @@
-const sql = require("@dataform/sql")();
+const sql = require("./sql")
 const segmentCommon = require("./common");
 
 module.exports = (params) => {
@@ -42,7 +42,7 @@ select
   *,
   coalesce(
     (
-      ${sql.timestamps.diff(`millisecond`,
+      ${sql.timestampDiff(`millisecond`,
       sql.windowFunction(
       "lag",
         "timestamp",

includes/sessions.js

@@ -1,5 +1,5 @@
 const segmentCommon = require("./common");
-const sql = require("@dataform/sql")();
+const sql = require("./sql")
 
 module.exports = (params) => {
 
@@ -92,7 +92,7 @@ select
   ${ctx.when(global.dataform.projectConfig.warehouse == "bigquery", `struct(\n  `)}
   ${segmentCommon.enabledEvents(params).map((event) => 
     `count(segment_sessionized_events.${event}_id) as total_${event}s`).join(`,\n  `)},
-  ${sql.timestamps.diff("millisecond", "min(segment_sessionized_events.timestamp)", "max(segment_sessionized_events.timestamp)")} as duration_millis
+  ${sql.timestampDiff("millisecond", "min(segment_sessionized_events.timestamp)", "max(segment_sessionized_events.timestamp)")} as duration_millis
   ${ctx.when(global.dataform.projectConfig.warehouse == "bigquery", `) as stats`)}
 
   -- first values in the session for page fields

includes/sql.js

@@ -0,0 +1,129 @@
+const getDialect = () => {
+  const dataformWarehouse = global.dataform.projectConfig.warehouse;
+  if (!dataformWarehouse) {
+    return "standard";
+  }
+  return {
+    bigquery: "standard",
+    redshift: "redshift",
+    postgres: "postgres",
+    snowflake: "snowflake",
+    sqldatawarehouse: "mssql",
+  }[dataformWarehouse];
+};
+
+const timestampDiff = (datePart, start, end) => {
+  const dialect = getDialect();
+  if (dialect === "snowflake" || dialect === "redshift") {
+    return `datediff(${datePart}, ${start}, ${end})`;
+  }
+  return `timestamp_diff(${end}, ${start}, ${datePart})`;
+};
+
+const windowFunction = (
+  name,
+  value,
+  ignoreNulls = false,
+  windowSpecification
+) => {
+  const dialect = getDialect();
+  const partitionFieldsAsString = windowSpecification.partitionFields
+    ? [...windowSpecification.partitionFields].join(`, `)
+    : "";
+  const orderFieldsAsString = windowSpecification.orderFields
+    ? [...windowSpecification.orderFields].join(`, `)
+    : "";
+
+  if (
+    dialect === "standard" ||
+    dialect === "mssql" ||
+    dialect === "snowflake"
+  ) {
+    return `${name}(${value} ${ignoreNulls ? `ignore nulls` : ``}) over (${
+      windowSpecification.partitionFields
+        ? `partition by ${partitionFieldsAsString}`
+        : ``
+    } ${
+      windowSpecification.orderFields ? `order by ${orderFieldsAsString}` : ``
+    } ${
+      windowSpecification.frameClause ? windowSpecification.frameClause : ``
+    })`;
+  }
+
+  // For some window functions in Redshift, a frame clause is always required
+  const requiresFrame = [
+    "avg",
+    "count",
+    "first_value",
+    "last_value",
+    "max",
+    "min",
+    "nth_value",
+    "stddev_samp",
+    "stddev_pop",
+    "stddev",
+    "sum",
+    "variance",
+    "var_samp",
+    "var_pop",
+  ].includes(name.toLowerCase());
+
+  if (dialect === "redshift") {
+    return `${name}(${value} ${ignoreNulls ? `ignore nulls` : ``}) over (${
+      windowSpecification.partitionFields
+        ? `partition by ${partitionFieldsAsString}`
+        : ``
+    } ${
+      windowSpecification.orderFields ? `order by ${orderFieldsAsString}` : ``
+    } ${
+      windowSpecification.orderFields
+        ? windowSpecification.frameClause
+          ? windowSpecification.frameClause
+          : requiresFrame
+          ? `rows between unbounded preceding and unbounded following`
+          : ``
+        : ``
+    })`;
+  }
+
+  if (dialect === "postgres") {
+    return `${name}(${value}) over (${
+      windowSpecification.partitionFields
+        ? `partition by ${partitionFieldsAsString}`
+        : ``
+    } ${windowSpecification.orderFields || ignoreNulls ? `order by` : ``} ${
+      ignoreNulls ? `case when ${value} is not null then 0 else 1 end asc` : ``
+    } ${orderFieldsAsString && ignoreNulls ? `,` : ``} ${orderFieldsAsString} ${
+      windowSpecification.orderFields
+        ? windowSpecification.frameClause
+          ? windowSpecification.frameClause
+          : requiresFrame
+          ? `rows between unbounded preceding and unbounded following`
+          : ``
+        : ``
+    })`;
+  }
+};
+
+const asString = (castableToString) => {
+  const dialect = getDialect();
+  if (dialect === "postgres" || dialect === "redshift") {
+    return `cast(${castableToString} as varchar)`;
+  }
+  return `cast(${castableToString} as string)`;
+};
+
+const surrogateKey = (columnNames) => {
+  const dialect = getDialect();
+  const columnsAsStrings = columnNames.map((id) => asString(id)).join(`,`);
+  if (dialect === "standard") {
+    return asString(`farm_fingerprint(concat(${columnsAsStrings}))`);
+  }
+  return asString(`md5(concat(${columnsAsStrings}))`);
+};
+
+module.exports = {
+  timestampDiff,
+  windowFunction,
+  surrogateKey,
+};

includes/user_map.js

@@ -1,4 +1,4 @@
-const sql = require("@dataform/sql")();
+const sql = require("./sql")
 const segmentCommon = require("./common");
 
 module.exports = (params) => {

includes/users.js

@@ -1,4 +1,4 @@
-const sql = require("@dataform/sql")();
+const sql = require("./sql")
 
 let USER = `coalesce(
   identifies.user_id,

package.json

@@ -1,7 +1,6 @@
 {
     "name": "dataform-segment",
     "dependencies": {
-        "@dataform/core": "1.15.5",
-        "@dataform/sql": "0.2.0"
+        "@dataform/core": "1.15.5"
     }
 }