Merge contributions manually (#2086)

* Add these changes from #2072 and #2074

* Need to also change the way arrays are unnested.

* mtoy's take on the changes ... passes ci-snowflake

* add fix and test for array names ending in numbers

* Add changes from #2088

* delete line for linter

---------

Co-authored-by: Michael Toy <66150587+mtoy-googly-moogly@users.noreply.github.com>

Author: lloydtabb

package.json

@@ -36,6 +36,7 @@
     "test-bigquery": "MALLOY_DATABASE=bigquery jest --runInBand",
     "test-postgres": "MALLOY_DATABASE=postgres jest --runInBand",
     "test-duckdb": "JEST_SILENT_REPORTER_SHOW_PATHS=true MALLOY_DATABASE=duckdb jest --runInBand --reporters jest-silent-reporter",
+    "ci-snowflake": "JEST_SILENT_REPORTER_SHOW_PATHS=true jest --config jest.snowflake.config.ts --reporters jest-silent-reporter --reporters summary",
     "test-silent": "JEST_SILENT_REPORTER_SHOW_PATHS=true jest --runInBand --reporters jest-silent-reporter --no-color",
     "test-deps": "npm run build && npx jest -t dependencies",
     "third-party-licenses": "ts-node scripts/third_party_licenses",

packages/malloy-db-snowflake/src/snowflake_connection.spec.ts

@@ -105,4 +105,17 @@ describe('db:Snowflake', () => {
     expect(res.rows[0]['rand']).toBeGreaterThanOrEqual(0);
     expect(res.rows[0]['rand']).toBeLessThanOrEqual(1);
   });
+
+  it('variant parser is not confused by arrays with numbers in name', async () => {
+    const x: malloy.SQLSourceDef = {
+      type: 'sql_select',
+      name: 'one_two_three',
+      connection: conn.name,
+      dialect: conn.dialectName,
+      selectStr: 'SELECT [1,2,3] as one_23',
+      fields: [],
+    };
+    const y = await conn.fetchSelectSchema(x);
+    expect(y.fields[0].name).toEqual('ONE_23');
+  });
 });

packages/malloy-db-snowflake/src/snowflake_connection.ts

@@ -63,6 +63,9 @@ export interface SnowflakeConnectionOptions {
   scratchSpace?: namespace;
 
   queryOptions?: RunSQLOptions;
+
+  // Timeout for the statement
+  timeoutMs?: number;
 }
 
 type PathChain =
@@ -207,6 +210,11 @@ class SnowArray extends SnowField {
   }
 }
 
+/**
+ * Default statement timeoutMs value, 10 Mins
+ */
+const TIMEOUT_MS = 1000 * 60 * 10;
+
 export class SnowflakeConnection
   extends BaseConnection
   implements
@@ -221,6 +229,7 @@ export class SnowflakeConnection
   // the database & schema where we do temporary operations like creating a temp table
   private scratchSpace?: namespace;
   private queryOptions: RunSQLOptions;
+  private timeoutMs: number;
 
   constructor(
     public readonly name: string,
@@ -235,6 +244,7 @@ export class SnowflakeConnection
     this.executor = new SnowflakeExecutor(connOptions, options?.poolOptions);
     this.scratchSpace = options?.scratchSpace;
     this.queryOptions = options?.queryOptions ?? {};
+    this.timeoutMs = options?.timeoutMs ?? TIMEOUT_MS;
   }
 
   get dialectName(): string {
@@ -273,10 +283,10 @@ export class SnowflakeConnection
 
   public async runSQL(
     sql: string,
-    options?: RunSQLOptions
+    options: RunSQLOptions = {}
   ): Promise<MalloyQueryData> {
     const rowLimit = options?.rowLimit ?? this.queryOptions?.rowLimit;
-    let rows = await this.executor.batch(sql);
+    let rows = await this.executor.batch(sql, options, this.timeoutMs);
     if (rowLimit !== undefined && rows.length > rowLimit) {
       rows = rows.slice(0, rowLimit);
     }
@@ -329,12 +339,24 @@ export class SnowflakeConnection
     }
     // For these things, we need to sample the data to know the schema
     if (variants.length > 0) {
+      // * remove null values
+      // * remove fields for which we have multiple types
+      //   ( requires folding decimal to integer )
       const sampleQuery = `
-      SELECT regexp_replace(PATH, '\\[[0-9]+\\]', '[*]') as PATH, lower(TYPEOF(value)) as type
-      FROM (select object_construct(*) o from  ${tablePath} limit 100)
-            ,table(flatten(input => o, recursive => true)) as meta
-      GROUP BY 1,2
-      ORDER BY PATH;
+        select path, min(type) as type
+        from (
+          select
+            regexp_replace(path, '\\\\[[0-9]+\\\\]', '[*]') as path,
+            case when typeof(value) = 'INTEGER' then 'decimal' else lower(typeof(value)) end as type
+          from
+            (select object_construct(*) o from ${tablePath} limit 100)
+              ,table(flatten(input => o, recursive => true)) as meta
+          group by 1,2
+        )
+        where type != 'null_value'
+        group BY 1
+        having count(*) <=1
+        order by path;
       `;
       const fieldPathRows = await this.executor.batch(sampleQuery);
 

packages/malloy-db-snowflake/src/snowflake_executor.ts

@@ -149,15 +149,30 @@ export class SnowflakeExecutor {
     });
   }
 
-  public async _execute(sqlText: string, conn: Connection): Promise<QueryData> {
-    return new Promise((resolve, reject) => {
-      const _statment = conn.execute({
+  public async _execute(
+    sqlText: string,
+    conn: Connection,
+    options?: RunSQLOptions,
+    timeoutMs?: number
+  ): Promise<QueryData> {
+    let _statement: RowStatement | undefined;
+    const cancel = () => {
+      _statement?.cancel();
+    };
+    const timeoutId = timeoutMs ? setTimeout(cancel, timeoutMs) : undefined;
+    options?.abortSignal?.addEventListener('abort', cancel);
+    return await new Promise((resolve, reject) => {
+      _statement = conn.execute({
         sqlText,
         complete: (
           err: SnowflakeError | undefined,
           _stmt: RowStatement,
           rows?: QueryData
         ) => {
+          options?.abortSignal?.removeEventListener('abort', cancel);
+          if (timeoutId) {
+            clearTimeout(timeoutId);
+          }
           if (err) {
             reject(err);
           } else if (rows) {
@@ -186,10 +201,14 @@ export class SnowflakeExecutor {
     );
   }
 
-  public async batch(sqlText: string): Promise<QueryData> {
+  public async batch(
+    sqlText: string,
+    options?: RunSQLOptions,
+    timeoutMs?: number
+  ): Promise<QueryData> {
     return await this.pool_.use(async (conn: Connection) => {
       await this._setSessionParams(conn);
-      return await this._execute(sqlText, conn);
+      return await this._execute(sqlText, conn, options, timeoutMs);
     });
   }
 

packages/malloy/src/dialect/snowflake/snowflake.ts

@@ -205,7 +205,7 @@ export class SnowflakeDialect extends Dialect {
   ): string {
     const as = this.sqlMaybeQuoteIdentifier(alias);
     if (isArray) {
-      return `,LATERAL FLATTEN(INPUT => ${source}) AS ${alias}_1, LATERAL (SELECT ${alias}_1.INDEX, object_construct('value', ${alias}_1.value) as value ) as ${as}`;
+      return `LEFT JOIN lateral flatten(input => ${source}) as ${as}`;
     } else {
       // have to have a non empty row or it treats it like an inner join :barf-emoji:
       return `LEFT JOIN LATERAL FLATTEN(INPUT => ifnull(${source},[1])) AS ${as}`;
@@ -263,11 +263,11 @@ export class SnowflakeDialect extends Dialect {
     const sqlName = this.sqlMaybeQuoteIdentifier(childName);
     if (childName === '__row_id') {
       return `"${parentAlias}".INDEX::varchar`;
-    } else if (
-      parentType === 'array[scalar]' ||
-      parentType === 'array[record]'
-    ) {
-      const arrayRef = `"${parentAlias}".value:${sqlName}`;
+    } else if (parentType.startsWith('array')) {
+      let arrayRef = `"${parentAlias}".value`;
+      if (parentType === 'array[record]') {
+        arrayRef += `:${sqlName}`;
+      }
       switch (childType) {
         case 'record':
         case 'array':