feat: add policy tags and masking policy support for maxcompute

plugins/extractors/maxcompute/client/client.go

@@ -118,3 +118,17 @@ func (c *Client) GetTablePreview(_ context.Context, partitionValue string, table
 
 	return columnNames, protoList, nil
 }
+
+func (_ *Client) GetPolicyTagsAndMaskingPolicy(table *odps.Table) (string, []string, error) {
+	var policyTags []string
+	var maskingPolicy string
+	colPolicyTags, err := table.ColumnMaskInfos()
+	if err != nil {
+		return maskingPolicy, nil, err
+	}
+	for _, policyTag := range colPolicyTags {
+		maskingPolicy = policyTag.Name
+		policyTags = append(policyTags, policyTag.PolicyNameList...)
+	}
+	return maskingPolicy, policyTags, nil
+}

plugins/extractors/maxcompute/maxcompute.go

@@ -80,6 +80,7 @@ type Client interface {
 	ListTable(ctx context.Context, schemaName string) ([]*odps.Table, error)
 	GetTableSchema(ctx context.Context, table *odps.Table) (string, *tableschema.TableSchema, error)
 	GetTablePreview(ctx context.Context, partitionValue string, table *odps.Table, maxRows int) ([]string, *structpb.ListValue, error)
+	GetPolicyTagsAndMaskingPolicy(table *odps.Table) (string, []string, error)
 }
 
 func New(logger log.Logger, clientFunc NewClientFunc, randFn randFn) *Extractor {
@@ -212,7 +213,7 @@ func (e *Extractor) buildAsset(ctx context.Context, schema *odps.Schema,
 		Service:     maxcomputeService,
 	}
 
-	tableAttributesData := e.buildTableAttributesData(schemaName, tableType, tableSchema)
+	tableAttributesData := e.buildTableAttributesData(schemaName, tableType, table, tableSchema)
 
 	if tableType == config.TableTypeView {
 		query := tableSchema.ViewText
@@ -298,7 +299,7 @@ func buildColumns(dataType datatype.DataType) []*v1beta2.Column {
 	return columns
 }
 
-func (e *Extractor) buildTableAttributesData(schemaName, tableType string, tableInfo *tableschema.TableSchema) map[string]interface{} {
+func (e *Extractor) buildTableAttributesData(schemaName, tableType string, table *odps.Table, tableInfo *tableschema.TableSchema) map[string]interface{} {
 	attributesData := map[string]interface{}{}
 
 	attributesData["project_name"] = e.config.ProjectName
@@ -321,6 +322,15 @@ func (e *Extractor) buildTableAttributesData(schemaName, tableType string, table
 		attributesData["partition_fields"] = partitionNames
 	}
 
+	maskingPolicy, policyTags, err := e.client.GetPolicyTagsAndMaskingPolicy(table)
+	if err != nil {
+		e.logger.Warn("error getting policy tags", "error", err)
+	}
+	attributesData["masking_policy"] = maskingPolicy
+	if len(policyTags) > 0 {
+		attributesData["policy_tags"] = policyTags
+	}
+
 	return attributesData
 }