seaweedfs/weed/worker/tasks/iceberg/detection.go

package iceberg

import (
	"context"
	"errors"
	"fmt"
	"path"
	"strings"
	"time"

	"github.com/apache/iceberg-go"
	"github.com/apache/iceberg-go/table"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
)

// tableInfo captures metadata about a table for detection/execution.
type tableInfo struct {
	BucketName       string
	Namespace        string
	TableName        string
	TablePath        string // namespace/tableName
	MetadataFileName string
	Metadata         table.Metadata
}

// scanTablesForMaintenance enumerates table buckets and their tables,
// evaluating which ones need maintenance based on metadata thresholds.
// When limit > 0 the scan stops after collecting limit+1 results so the
// caller can determine whether more tables remain (HasMore).
func (h *Handler) scanTablesForMaintenance(
	ctx context.Context,
	filerClient filer_pb.SeaweedFilerClient,
	config Config,
	bucketFilter, namespaceFilter, tableFilter string,
	limit int,
) ([]tableInfo, error) {
	var tables []tableInfo
	ops, err := parseOperations(config.Operations)
	if err != nil {
		return nil, fmt.Errorf("parse operations: %w", err)
	}

	// Compile wildcard matchers once (nil = match all)
	bucketMatchers := wildcard.CompileWildcardMatchers(bucketFilter)
	nsMatchers := wildcard.CompileWildcardMatchers(namespaceFilter)
	tableMatchers := wildcard.CompileWildcardMatchers(tableFilter)

	bucketsPath := s3tables.TablesPath
	bucketEntries, err := listFilerEntries(ctx, filerClient, bucketsPath, "")
	if err != nil {
		return nil, fmt.Errorf("list buckets: %w", err)
	}

	for _, bucketEntry := range bucketEntries {
		select {
		case <-ctx.Done():
			return tables, ctx.Err()
		default:
		}

		if !bucketEntry.IsDirectory || !s3tables.IsTableBucketEntry(bucketEntry) {
			continue
		}
		bucketName := bucketEntry.Name
		if !wildcard.MatchesAnyWildcard(bucketMatchers, bucketName) {
			continue
		}
		bucketMaintenance, err := parseMaintenanceConfiguration(bucketEntry.Extended, bucketName)
		if err != nil {
			glog.Warningf("iceberg maintenance: skipping bucket %s: %v", bucketName, err)
			continue
		}

		// List namespaces within the bucket
		bucketPath := path.Join(bucketsPath, bucketName)
		var recordRoots map[string][]string
		nsEntries, err := listFilerEntries(ctx, filerClient, bucketPath, "")
		if err != nil {
			glog.Warningf("iceberg maintenance: failed to list namespaces in bucket %s: %v", bucketName, err)
			continue
		}

		for _, nsEntry := range nsEntries {
			select {
			case <-ctx.Done():
				return tables, ctx.Err()
			default:
			}

			if !nsEntry.IsDirectory {
				continue
			}
			nsName := nsEntry.Name
			if !wildcard.MatchesAnyWildcard(nsMatchers, nsName) {
				continue
			}
			// Skip internal directories
			if strings.HasPrefix(nsName, ".") {
				continue
			}

			// List tables within the namespace
			nsPath := path.Join(bucketPath, nsName)
			tableEntries, err := listFilerEntries(ctx, filerClient, nsPath, "")
			if err != nil {
				glog.Warningf("iceberg maintenance: failed to list tables in %s/%s: %v", bucketName, nsName, err)
				continue
			}

			for _, tableEntry := range tableEntries {
				if !tableEntry.IsDirectory {
					continue
				}
				tblName := tableEntry.Name
				if !wildcard.MatchesAnyWildcard(tableMatchers, tblName) {
					continue
				}

				// Check if this entry has table metadata
				metadataBytes, ok := tableEntry.Extended[s3tables.ExtendedKeyMetadata]
				if !ok || len(metadataBytes) == 0 {
					continue
				}

				tablePath := path.Join(nsName, tblName)
				state, err := parseTableMetadataEnvelope(metadataBytes, bucketName, tablePath)
				if err != nil {
					if errors.Is(err, errForeignFormat) {
						glog.V(3).Infof("iceberg maintenance: skipping %s/%s/%s: %v", bucketName, nsName, tblName, err)
					} else {
						glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot parse iceberg metadata: %v", bucketName, nsName, tblName, err)
					}
					continue
				}
				// Maintenance deletes what a table's metadata does not reference,
				// so a table sharing its files with another record is left alone.
				if recordRoots == nil {
					if recordRoots, err = s3tables.CatalogRecordRoots(ctx, filerClient, bucketName); err != nil {
						glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot list catalog records: %v", bucketName, nsName, tblName, err)
						continue
					}
				}
				if filesShared(recordRoots, path.Join(bucketPath, tablePath), path.Join(bucketPath, state.DataPath)) {
					glog.V(1).Infof("iceberg maintenance: skipping %s/%s/%s: its location is shared with another table or view", bucketName, nsName, tblName)
					continue
				}
				if err := state.loadMetadata(ctx, filerClient, bucketName); err != nil {
					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot load iceberg metadata: %v", bucketName, nsName, tblName, err)
					continue
				}
				if !isIcebergTableEntry(tableEntry.Extended, state.Metadata) {
					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: not an iceberg table", bucketName, nsName, tblName)
					continue
				}

				tableMaintenance, err := parseMaintenanceConfiguration(tableEntry.Extended, path.Join(bucketName, tablePath))
				if err != nil {
					glog.Warningf("iceberg maintenance: skipping %s/%s/%s: %v", bucketName, nsName, tblName, err)
					continue
				}
				maintenance := mergeMaintenanceConfiguration(bucketMaintenance, tableMaintenance)
				tableOps := filterDisabledOperations(ops, maintenance)
				if len(tableOps) == 0 {
					continue
				}
				effective := resolveTableConfig(config, state.Metadata.Properties(), maintenance)

				// A table whose files cannot be read is skipped for this pass;
				// the other tables are still evaluated.
				needsWork, err := guardRead(path.Join(bucketName, tablePath), func() (bool, error) {
					return h.tableNeedsMaintenance(ctx, filerClient, bucketName, tablePath, state, effective, tableOps)
				})
				if err != nil {
					glog.V(2).Infof("iceberg maintenance: skipping %s/%s/%s: cannot evaluate maintenance need: %v", bucketName, nsName, tblName, err)
					continue
				}
				if needsWork {
					tables = append(tables, tableInfo{
						BucketName:       bucketName,
						Namespace:        nsName,
						TableName:        tblName,
						TablePath:        tablePath,
						MetadataFileName: state.MetadataFileName,
						Metadata:         state.Metadata,
					})
					if limit > 0 && len(tables) > limit {
						return tables, nil
					}
				}
			}
		}
	}

	return tables, nil
}

// filesShared reports whether a record other than selfPath keeps its files at
// root, above it or under it. A record that only sits on root, as one created
// over a renamed table's old name does, keeps no files there.
func filesShared(roots map[string][]string, selfPath, root string) bool {
	for recordPath, others := range roots {
		if recordPath == selfPath {
			continue
		}
		for _, other := range others {
			if s3tables.PathsOverlap(other, root) {
				return true
			}
		}
	}
	return false
}

func normalizeDetectionConfig(config Config) Config {
	config = applyThresholdDefaults(config)
	if config.SnapshotRetentionMs <= 0 {
		config.SnapshotRetentionMs = hoursToMs(defaultSnapshotRetentionHours)
	}
	if config.MaxSnapshotsToKeep <= 0 {
		config.MaxSnapshotsToKeep = defaultMaxSnapshotsToKeep
	}
	return config
}

func (h *Handler) tableNeedsMaintenance(
	ctx context.Context,
	filerClient filer_pb.SeaweedFilerClient,
	bucketName, tablePath string,
	state *tableState,
	config Config,
	ops []string,
) (bool, error) {
	config = normalizeDetectionConfig(config)
	meta := state.Metadata
	dataPath := state.DataPath
	metadataFileName := state.MetadataFileName
	cachedPlanningIndex := state.PlanningIndex

	var predicate *partitionPredicate
	if strings.TrimSpace(config.Where) != "" {
		needsPredicate := false
		for _, op := range ops {
			if op == "compact" || op == "rewrite_position_delete_files" || op == "rewrite_manifests" {
				needsPredicate = true
				break
			}
		}
		if needsPredicate {
			var err error
			predicate, err = parsePartitionPredicate(config.Where, meta)
			if err != nil {
				return false, err
			}
		}
	}
	_ = predicate // used by rewrite_position_delete_files; planning index handles compact/rewrite_manifests

	// Evaluate the metadata-only expiration check first so large tables do not
	// pay for manifest reads when snapshot expiry already makes them eligible.
	for _, op := range ops {
		if op == "expire_snapshots" && needsMaintenance(meta, config) {
			return true, nil
		}
	}

	var currentManifests []iceberg.ManifestFile
	var manifestsErr error
	var manifestsLoaded bool
	getCurrentManifests := func() ([]iceberg.ManifestFile, error) {
		if manifestsLoaded {
			return currentManifests, manifestsErr
		}
		currentManifests, manifestsErr = loadCurrentManifests(ctx, filerClient, bucketName, dataPath, meta)
		manifestsLoaded = true
		return currentManifests, manifestsErr
	}

	computedPlanningIndexes := make(map[string]*planningIndex)
	planningIndexLoaded := make(map[string]bool)
	planningIndexErrs := make(map[string]error)
	getPlanningIndex := func(op string) (*planningIndex, error) {
		if planningIndexLoaded[op] {
			return computedPlanningIndexes[op], planningIndexErrs[op]
		}
		planningIndexLoaded[op] = true

		manifests, err := getCurrentManifests()
		if err != nil {
			planningIndexErrs[op] = err
			return nil, err
		}
		index, err := buildPlanningIndexFromManifests(ctx, filerClient, bucketName, dataPath, meta, config, []string{op}, manifests)
		if err != nil {
			planningIndexErrs[op] = err
			return nil, err
		}
		computedPlanningIndexes[op] = index
		if index != nil {
			if err := persistPlanningIndex(ctx, filerClient, bucketName, tablePath, index); err != nil {
				glog.V(2).Infof("iceberg maintenance: unable to persist planning index for %s/%s: %v", bucketName, tablePath, err)
			}
		}
		return index, nil
	}
	checkPlanningIndex := func(op string, eligibleFn func(*planningIndex, Config) (bool, bool)) (bool, error) {
		if cachedPlanningIndex != nil && cachedPlanningIndex.matchesSnapshot(meta) {
			if eligible, ok := eligibleFn(cachedPlanningIndex, config); ok {
				return eligible, nil
			}
		}

		index, err := getPlanningIndex(op)
		if err != nil {
			return false, err
		}
		if index == nil {
			return false, nil
		}

		eligible, _ := eligibleFn(index, config)
		return eligible, nil
	}

	var opEvalErrors []string
	planningIndexErrorReported := false

	for _, op := range ops {
		switch op {
		case "expire_snapshots":
			// Handled by the metadata-only check above.
			continue
		case "compact":
			eligible, err := checkPlanningIndex(op, (*planningIndex).compactionEligible)
			if err != nil {
				if !planningIndexErrorReported {
					opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
					planningIndexErrorReported = true
				}
				continue
			}
			if eligible {
				return true, nil
			}
		case "rewrite_position_delete_files":
			if hasRowLineage(meta.Version()) {
				continue // the rewrite would add position-delete files, which a v3 table must not get (skipDeleteRewriteV3)
			}
			manifests, err := getCurrentManifests()
			if err != nil {
				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
				continue
			}
			eligible, err := hasEligibleDeleteRewrite(ctx, filerClient, bucketName, dataPath, manifests, config, meta, predicate)
			if err != nil {
				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
				continue
			}
			if eligible {
				return true, nil
			}
		case "rewrite_manifests":
			eligible, err := checkPlanningIndex(op, (*planningIndex).rewriteManifestsEligible)
			if err != nil {
				if !planningIndexErrorReported {
					opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
					planningIndexErrorReported = true
				}
				continue
			}
			if eligible {
				return true, nil
			}
		case "remove_orphans":
			if metadataFileName == "" {
				currentState, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
				if err != nil {
					opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
					continue
				}
				metadataFileName = currentState.MetadataFileName
			}
			orphanCandidates, err := collectOrphanCandidates(ctx, filerClient, bucketName, dataPath, meta, metadataFileName, config.OrphanOlderThanHours)
			if err != nil {
				opEvalErrors = append(opEvalErrors, fmt.Sprintf("%s: %v", op, err))
				continue
			}
			if len(orphanCandidates) > 0 {
				return true, nil
			}
		}
	}

	if len(opEvalErrors) > 0 {
		return false, fmt.Errorf("evaluate maintenance operations: %s", strings.Join(opEvalErrors, "; "))
	}

	return false, nil
}

func metadataFileNameFromLocation(location string) string {
	if location == "" {
		return ""
	}
	name := path.Base(location)
	if name == "." || name == "/" {
		return ""
	}
	return name
}

func countDataManifests(manifests []iceberg.ManifestFile) int64 {
	var count int64
	for _, mf := range manifests {
		if mf.ManifestContent() == iceberg.ManifestContentData {
			count++
		}
	}
	return count
}

func loadCurrentManifests(
	ctx context.Context,
	filerClient filer_pb.SeaweedFilerClient,
	bucketName, dataPath string,
	meta table.Metadata,
) ([]iceberg.ManifestFile, error) {
	currentSnap := meta.CurrentSnapshot()
	if currentSnap == nil || currentSnap.ManifestList == "" {
		return nil, nil
	}

	manifestListData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, currentSnap.ManifestList)
	if err != nil {
		return nil, fmt.Errorf("read manifest list: %w", err)
	}
	manifests, err := s3tables.ReadManifestList(manifestListData)
	if err != nil {
		return nil, fmt.Errorf("parse manifest list: %w", err)
	}
	return manifests, nil
}

func hasEligibleCompaction(
	ctx context.Context,
	filerClient filer_pb.SeaweedFilerClient,
	bucketName, dataPath string,
	manifests []iceberg.ManifestFile,
	config Config,
	meta table.Metadata,
	predicate *partitionPredicate,
) (bool, error) {
	if len(manifests) == 0 {
		return false, nil
	}

	minInputFiles, err := compactionMinInputFiles(config.MinInputFiles)
	if err != nil {
		return false, err
	}

	var dataManifests []iceberg.ManifestFile
	specIDs := make(map[int32]struct{})
	for _, mf := range manifests {
		if mf.ManifestContent() != iceberg.ManifestContentData {
			continue
		}
		dataManifests = append(dataManifests, mf)
		specIDs[mf.PartitionSpecID()] = struct{}{}
	}
	if len(dataManifests) == 0 {
		return false, nil
	}
	if len(specIDs) > 1 {
		return false, nil
	}

	specsByID := specByID(meta)
	schema := meta.CurrentSchema()

	var allEntries []iceberg.ManifestEntry
	var dvPaths map[string]bool
	for _, mf := range dataManifests {
		manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
		if err != nil {
			return false, fmt.Errorf("read manifest %s: %w", mf.FilePath(), err)
		}
		entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema)
		if err != nil {
			return false, fmt.Errorf("parse manifest %s: %w", mf.FilePath(), err)
		}
		allEntries = append(allEntries, entries...)
	}

	// Deletion vectors exist only from format version 3.
	if hasRowLineage(meta.Version()) {
		var deleteEntries []iceberg.ManifestEntry
		for _, mf := range manifests {
			if mf.ManifestContent() != iceberg.ManifestContentDeletes {
				continue
			}
			manifestData, err := loadFileByIcebergPath(ctx, filerClient, bucketName, dataPath, mf.FilePath())
			if err != nil {
				return false, fmt.Errorf("read delete manifest %s: %w", mf.FilePath(), err)
			}
			entries, err := s3tables.ReadManifest(mf, manifestData, true, specsByID, schema)
			if err != nil {
				return false, fmt.Errorf("parse delete manifest %s: %w", mf.FilePath(), err)
			}
			deleteEntries = append(deleteEntries, entries...)
		}
		if dvPaths, err = deletionVectorPaths(deleteEntries, bucketName, dataPath); err != nil {
			return false, err
		}
	}

	candidateEntries, err := excludeDeletionVectorFiles(allEntries, dvPaths, bucketName, dataPath)
	if err != nil {
		return false, err
	}
	if predicate != nil {
		filtered := make([]iceberg.ManifestEntry, 0, len(candidateEntries))
		for _, entry := range candidateEntries {
			spec, ok := specsByID[int(entry.DataFile().SpecID())]
			if !ok {
				continue
			}
			match, err := predicate.Matches(spec, entry.DataFile().Partition())
			if err != nil {
				return false, err
			}
			if match {
				filtered = append(filtered, entry)
			}
		}
		candidateEntries = filtered
	}

	rewritePlan, err := resolveCompactionRewritePlan(config, meta)
	if err != nil {
		return false, fmt.Errorf("resolve rewrite strategy: %w", err)
	}

	targetSize := compactionTargetSizeForPlan(config, rewritePlan)
	bins := buildCompactionBins(candidateEntries, targetSize, minInputFiles, meta)
	bins = filterCompactionBinsByPlan(bins, config, rewritePlan)
	return len(bins) > 0, nil
}

func compactionMinInputFiles(minInputFiles int64) (int, error) {
	// Ensure the configured value is positive and fits into the platform's int type
	if minInputFiles <= 0 {
		return 0, fmt.Errorf("min input files must be positive, got %d", minInputFiles)
	}
	maxInt := int64(^uint(0) >> 1)
	if minInputFiles > maxInt {
		return 0, fmt.Errorf("min input files %d exceeds platform int size", minInputFiles)
	}
	return int(minInputFiles), nil
}

// needsMaintenance checks whether snapshot expiration work is needed based on
// metadata-only thresholds.
// It asks execution what it would do rather than reimplementing the rules:
// expiry always requires a snapshot to be past the retention window, so a
// table over the snapshot quota whose snapshots are all young, or all pinned by
// refs, would otherwise be proposed for a job that can only no-op.
func needsMaintenance(meta table.Metadata, config Config) bool {
	return len(snapshotsToExpire(meta, config, time.Now().UnixMilli())) > 0
}

// buildMaintenanceProposal creates a JobProposal for a table needing maintenance.
func (h *Handler) buildMaintenanceProposal(t tableInfo, filerGrpcAddress, resourceGroup string) *plugin_pb.JobProposal {
	dedupeKey := fmt.Sprintf("iceberg_maintenance:%s/%s/%s", t.BucketName, t.Namespace, t.TableName)

	snapshotCount := len(t.Metadata.Snapshots())
	summary := fmt.Sprintf("Maintain %s/%s/%s (%d snapshots)", t.BucketName, t.Namespace, t.TableName, snapshotCount)

	proposal := &plugin_pb.JobProposal{
		ProposalId: fmt.Sprintf("iceberg-%s-%s-%s-%d", t.BucketName, t.Namespace, t.TableName, time.Now().UnixMilli()),
		DedupeKey:  dedupeKey,
		JobType:    jobType,
		Priority:   plugin_pb.JobPriority_JOB_PRIORITY_NORMAL,
		Summary:    summary,
		Parameters: map[string]*plugin_pb.ConfigValue{
			"bucket_name":   {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: t.BucketName}},
			"namespace":     {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: t.Namespace}},
			"table_name":    {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: t.TableName}},
			"table_path":    {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: t.TablePath}},
			"filer_address": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: filerGrpcAddress}},
		},
		Labels: map[string]string{
			"bucket":    t.BucketName,
			"namespace": t.Namespace,
			"table":     t.TableName,
		},
	}
	if resourceGroup != "" {
		proposal.Parameters["resource_group"] = &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_StringValue{StringValue: resourceGroup}}
		proposal.Labels["resource_group"] = resourceGroup
	}
	return proposal
}