feat(bigquery): add label-based table exclusion

Tables matching any configured exclude label (key-value pair) are
skipped during extraction. The check runs after table metadata is
fetched since labels are part of TableMetadata.

Closes #460

Author: ravisuhag

plugins/extractors/bigquery/README.md

@@ -17,6 +17,8 @@ source:
         - dataset_b
       tables:
         - dataset_c.table_a
+      labels:
+        env: staging
     max_page_size: 100
     include_column_profile: true
     build_view_lineage: true
@@ -53,6 +55,7 @@ source:
 | `table_pattern` | `string` | `gofood.fact_` | Regex pattern to whitelist tables to extract | _optional_ |
 | `exclude.datasets` | `[]string` | `[dataset_a]` | Dataset IDs to exclude | _optional_ |
 | `exclude.tables` | `[]string` | `[dataset_c.table_a]` | Table names in `datasetID.tableID` format to exclude | _optional_ |
+| `exclude.labels` | `map[string]string` | `{env: staging}` | Tables with any matching label key-value pair are excluded | _optional_ |
 | `max_page_size` | `int` | `100` | Page size hint for BigQuery API list calls | _optional_ |
 | `dataset_page_size` | `int` | `10` | Page size for listing datasets. Falls back to `max_page_size` | _optional_ |
 | `table_page_size` | `int` | `50` | Page size for listing tables. Falls back to `max_page_size` | _optional_ |

plugins/extractors/bigquery/bigquery.go

@@ -62,6 +62,8 @@ type Exclude struct {
 	Datasets []string `json:"datasets" yaml:"datasets" mapstructure:"datasets"`
 	// list of tableNames in format - datasetID.tableID
 	Tables []string `json:"tables" yaml:"tables" mapstructure:"tables"`
+	// list of label key-value pairs; tables matching any label are excluded
+	Labels map[string]string `json:"labels" yaml:"labels" mapstructure:"labels"`
 }
 
 const (
@@ -83,6 +85,8 @@ exclude:
 	- dataset_b
   tables:
 	- dataset_c.table_a
+  labels:
+	env: staging
 max_page_size: 100
 include_column_profile: true
 build_view_lineage: true
@@ -334,6 +338,14 @@ func (e *Extractor) extractTable(ctx context.Context, ds *bigquery.Dataset, emit
 					e.logger.Error("failed to fetch table metadata", "err", err, "table", tableFQN)
 					return nil
 				}
+				if IsExcludedByLabels(tmd.Labels, e.config.Exclude.Labels) {
+					e.excludedTableCtr.Add(ctx, 1, metric.WithAttributes(
+						attribute.String("bq.project_id", e.config.ProjectID),
+						attribute.String("bq.dataset_id", ds.DatasetID),
+					))
+					e.logger.Debug("excluding table by labels", "dataset_id", ds.DatasetID, "table_id", table.TableID)
+					return nil
+				}
 				record, err := e.buildRecord(ctx, table, tmd)
 				if err != nil {
 					e.logger.Error("failed to build record", "err", err, "table", tableFQN)
@@ -799,6 +811,21 @@ func IsExcludedTable(datasetID, tableID string, excludedTables []string) bool {
 	return false
 }
 
+// isExcludedByLabels returns true if the table's labels match any of the
+// configured exclude labels. A match means the table has the same key with
+// the same value.
+func IsExcludedByLabels(tableLabels, excludeLabels map[string]string) bool {
+	if len(excludeLabels) == 0 {
+		return false
+	}
+	for k, v := range excludeLabels {
+		if tableLabels[k] == v {
+			return true
+		}
+	}
+	return false
+}
+
 // getMaxPageSize returns max_page_size if configured in recipe, otherwise returns default value
 func (e *Extractor) getMaxPageSize() int {
 	if e.config.MaxPageSize > 0 {

plugins/extractors/bigquery/exclude_test.go

@@ -0,0 +1,40 @@
+package bigquery
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsExcludedByLabels(t *testing.T) {
+	t.Run("should return false when no exclude labels configured", func(t *testing.T) {
+		tableLabels := map[string]string{"env": "prod"}
+		assert.False(t, IsExcludedByLabels(tableLabels, nil))
+		assert.False(t, IsExcludedByLabels(tableLabels, map[string]string{}))
+	})
+
+	t.Run("should return true when table has a matching label", func(t *testing.T) {
+		tableLabels := map[string]string{"env": "staging", "team": "data"}
+		excludeLabels := map[string]string{"env": "staging"}
+		assert.True(t, IsExcludedByLabels(tableLabels, excludeLabels))
+	})
+
+	t.Run("should return false when label key matches but value differs", func(t *testing.T) {
+		tableLabels := map[string]string{"env": "prod"}
+		excludeLabels := map[string]string{"env": "staging"}
+		assert.False(t, IsExcludedByLabels(tableLabels, excludeLabels))
+	})
+
+	t.Run("should return false when table has no labels", func(t *testing.T) {
+		excludeLabels := map[string]string{"env": "staging"}
+		assert.False(t, IsExcludedByLabels(nil, excludeLabels))
+		assert.False(t, IsExcludedByLabels(map[string]string{}, excludeLabels))
+	})
+
+	t.Run("should match any exclude label not all", func(t *testing.T) {
+		tableLabels := map[string]string{"env": "staging"}
+		excludeLabels := map[string]string{"env": "staging", "lifecycle": "ephemeral"}
+		// Matches "env: staging" even though table doesn't have "lifecycle".
+		assert.True(t, IsExcludedByLabels(tableLabels, excludeLabels))
+	})
+}