feat(warehouse): added support for warehouse column count limit #2723
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -119,6 +119,7 @@ type UploadJobT struct { | |
| hasAllTablesSkipped bool | ||
| tableUploadStatuses []*TableUploadStatusT | ||
| destinationValidator validations.DestinationValidator | ||
| stats stats.Stats | ||
| } | ||
|
|
||
| type UploadColumnT struct { | ||
|
|
@@ -145,8 +146,8 @@ var ( | |
| ) | ||
|
|
||
| var ( | ||
| maxParallelLoads map[string]int | ||
| columnCountLimitMap map[string]int | ||
| ) | ||
|
|
||
| func Init() { | ||
|
|
@@ -167,15 +168,15 @@ func setMaxParallelLoads() { | |
| warehouseutils.CLICKHOUSE: config.GetInt("Warehouse.clickhouse.maxParallelLoads", 3), | ||
| warehouseutils.DELTALAKE: config.GetInt("Warehouse.deltalake.maxParallelLoads", 3), | ||
| } | ||
|
There was a problem hiding this comment. We are not planning to add this for SNOWFLAKE because in case of SNOWFLAKE we have a limit on the row size to be 16MB and this will most likely not meet because of the payload max size we have. |
||
| columnCountLimitMap = map[string]int{ | ||
| warehouseutils.AZURE_SYNAPSE: config.GetInt("Warehouse.azure_synapse.columnCountLimit", 1024), | ||
| warehouseutils.BQ: config.GetInt("Warehouse.bigquery.columnCountLimit", 10000), | ||
| warehouseutils.CLICKHOUSE: config.GetInt("Warehouse.clickhouse.columnCountLimit", 1000), | ||
| warehouseutils.MSSQL: config.GetInt("Warehouse.mssql.columnCountLimit", 1024), | ||
| warehouseutils.POSTGRES: config.GetInt("Warehouse.postgres.columnCountLimit", 1600), | ||
| warehouseutils.RS: config.GetInt("Warehouse.redshift.columnCountLimit", 1600), | ||
| warehouseutils.SNOWFLAKE: config.GetInt("Warehouse.snowflake.columnCountLimit", 5000), | ||
| warehouseutils.S3_DATALAKE: config.GetInt("Warehouse.s3_datalake.columnCountLimit", 10000), | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -204,7 +205,7 @@ func (job *UploadJobT) trackLongRunningUpload() chan struct{} { | |
| case <-time.After(longRunningUploadStatThresholdInMin): | ||
| pkgLogger.Infof("[WH]: Registering stat for long running upload: %d, dest: %s", job.upload.ID, job.warehouse.Identifier) | ||
|
|
||
| job.stats.NewTaggedStat( | ||
| "warehouse.long_running_upload", | ||
| stats.CountType, | ||
| stats.Tags{ | ||
|
|
@@ -1067,11 +1068,31 @@ func (job *UploadJobT) loadTable(tName string) (alteredSchema bool, err error) { | |
| job.recordTableLoad(tName, numEvents) | ||
| } | ||
|
|
||
| job.columnCountStat(tName) | ||
|
|
||
| return | ||
| } | ||
|
|
||
| func (job *UploadJobT) columnCountStat(tableName string) { | ||
| var ( | ||
| columnCountLimit int | ||
| ok bool | ||
| ) | ||
| if columnCountLimit, ok = columnCountLimitMap[job.warehouse.Type]; !ok { | ||
| return | ||
| } | ||
| currentColumnsCount := len(job.schemaHandle.schemaInWarehouse[tableName]) | ||
| if currentColumnsCount > int(float64(columnCountLimit)*columnCountLimitThreshold) { | ||
| tags := []tag{ | ||
| { | ||
| name: "tableName", value: strings.ToLower(tableName), | ||
| }, | ||
| { | ||
| name: "columnCountLimit", value: strconv.Itoa(columnCountLimit), | ||
| }, | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. Would it make sense to report two metrics?
And instead of computing the threshold here, we compute it on the alert manager, by doing a similar operation There was a problem hiding this comment. Should we add other identifiers as well, like workspaceId and destinationId ? There was a problem hiding this comment. Can you elaborate the motivation behind computing it at the alert manager rather than here ? @lvrach There was a problem hiding this comment.
We are already adding these inside There was a problem hiding this comment.
There was a problem hiding this comment.
Since the threshold dictates whether an alert should be sent, it is more natural for me to have it on the alert manager. It makes it easier to change it. Also, the fact that we add
If you want to visualise the limit as a line in grafana it is easier if it's metric. There was a problem hiding this comment.
Good point. There was a problem hiding this comment.
I still feel this information is not much useful. Recording in separate stat and adding tags warehouseId, tableName for an almost constant value (changes only on destination type) would be tedious. Adds computational cost to kapacitor or prometheus executor(future) too. There was a problem hiding this comment. Better for debugging and visualization. |
||
| job.counterStat(`warehouse_load_table_column_count`, tags...).Count(currentColumnsCount) | ||
| } | ||
| return | ||
| } | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Removed these 2 tests because it does not test any behavior.