Do not use slow query from INFORMATION_SCHEMA (PeerDB-io#3573)

When the `INFORMATION_SCHEMA` table baloons it will lead to very slow
query execution. In our case - the queries end up taking ~5mins,
which drastically slow down the ingest.

The proposal in this PR is to swap to using a `SHOW COLUMNS IN TABLE...`
and then querying the result from that.
There is a Snowflake documentation link which highlights this problem, and solution as well,
https://community.snowflake.com/s/article/Select-the-list-of-columns-in-the-table-without-using-information-schema

There is an alternative to continue using the `INFORMATION_SCHEMA`
but have a cache for all the tables in the sync, I however, think that will be trickier to implement.

Author: alexstoick

flow/connectors/snowflake/qrep.go

@@ -10,6 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/snowflakedb/gosnowflake"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -182,14 +183,22 @@ func (c *SnowflakeConnector) getColsFromTable(ctx context.Context, tableName str
 		return nil, fmt.Errorf("failed to parse table name: %w", err)
 	}
 
-	rows, err := c.QueryContext(
-		ctx,
-		getTableSchemaSQL,
-		strings.ToUpper(schemaTable.Schema),
-		strings.ToUpper(schemaTable.Table),
-	)
-	if err != nil {
-		return nil, fmt.Errorf("failed to execute query: %w", err)
+	fq := fmt.Sprintf("%s.%s", strings.ToUpper(schemaTable.Schema), strings.ToUpper(schemaTable.Table))
+
+	channel := make(chan string, 1)
+	ctxWithOpt := gosnowflake.WithQueryIDChan(ctx, channel)
+	rows, err := c.QueryContext(ctxWithOpt, getTableColumnListSQL, fq)
+	if err != nil || rows.Err() != nil {
+		return nil, fmt.Errorf("failed to run getTableColumnList query: %w", err)
+	}
+	defer rows.Close()
+
+	qid := <-channel
+
+	rows, err = c.QueryContext(ctx, getTableSchemaSQL, qid)
+
+	if err != nil || rows.Err() != nil {
+		return nil, fmt.Errorf("failed to run getTableSchema query: %w", err)
 	}
 	defer rows.Close()
 
@@ -200,9 +209,13 @@ func (c *SnowflakeConnector) getColsFromTable(ctx context.Context, tableName str
 		if err := rows.Scan(&colName, &colType, &numericPrecision, &numericScale); err != nil {
 			return nil, fmt.Errorf("failed to scan row: %w", err)
 		}
+		parsedColType := colType.String
+		if colType.String == "FIXED" {
+			parsedColType = "NUMBER"
+		}
 		cols = append(cols, SnowflakeTableColumn{
 			ColumnName:       colName.String,
-			ColumnType:       colType.String,
+			ColumnType:       parsedColType,
 			NumericPrecision: numericPrecision.Int32,
 			NumericScale:     numericScale.Int32,
 		})

flow/connectors/snowflake/snowflake.go

@@ -60,9 +60,12 @@ const (
 	 ARRAY_AGG(DISTINCT _PEERDB_UNCHANGED_TOAST_COLUMNS) FROM %s.%s WHERE
 	 _PEERDB_BATCH_ID = %d AND _PEERDB_RECORD_TYPE != 2
 	 GROUP BY _PEERDB_DESTINATION_TABLE_NAME`
-	getTableSchemaSQL = `SELECT COLUMN_NAME, DATA_TYPE, NUMERIC_PRECISION, NUMERIC_SCALE FROM INFORMATION_SCHEMA.COLUMNS
-	 WHERE UPPER(TABLE_SCHEMA)=? AND UPPER(TABLE_NAME)=? ORDER BY ORDINAL_POSITION`
-
+	getTableColumnListSQL = `SHOW COLUMNS IN TABLE IDENTIFIER(?)`
+	getTableSchemaSQL     = `SELECT "column_name",
+		parse_json("data_type"):type::string AS type,
+		parse_json("data_type"):precision::string AS precision,
+		parse_json("data_type"):scale::string AS scale
+		FROM table(result_scan(?))`
 	checkIfTableExistsSQL = `SELECT TO_BOOLEAN(COUNT(1)) FROM INFORMATION_SCHEMA.TABLES
 	 WHERE TABLE_SCHEMA=? and TABLE_NAME=?`
 	dropTableIfExistsSQL = "DROP TABLE IF EXISTS %s.%s"