feat(confgen): parallelize per-sheet row parsing with bounded worker pool

internal/confgen/parser.go

@@ -8,6 +8,8 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"buf.build/go/protovalidate"
 	"github.com/tableauio/tableau/format"
@@ -18,6 +20,7 @@ import (
 	"github.com/tableauio/tableau/internal/x/xerrors"
 	"github.com/tableauio/tableau/internal/x/xfs"
 	"github.com/tableauio/tableau/internal/x/xproto"
+	"github.com/tableauio/tableau/log"
 	"github.com/tableauio/tableau/options"
 	"github.com/tableauio/tableau/proto/tableaupb"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -68,25 +71,35 @@ func (x *sheetExporter) ScatterAndExport(info *SheetInfo,
 		return filename
 	}
 	// parse main sheet
+	mainParseBegin := time.Now()
 	mainMsg, err := parseMessageFromOneImporter(info, x.collector, mainImpInfo)
 	if err != nil {
 		return err
 	}
+	var parseNanos atomic.Int64
+	var exportNanos atomic.Int64
+	parseNanos.Add(time.Since(mainParseBegin).Nanoseconds())
 	mainName := getExportedConfName(info, mainImpInfo)
+	mainExportBegin := time.Now()
 	err = storeMessage(mainMsg, mainName, info.LocationName, x.OutputDir, x.OutputOpt, x.validator)
 	if err != nil {
 		return err
 	}
+	exportNanos.Add(time.Since(mainExportBegin).Nanoseconds())
 
 	g := x.collector.NewGroup(context.Background())
 	for _, impInfo := range impInfos {
 		// map-reduce: map jobs for concurrent processing
 		g.Go(func(ctx context.Context) error {
+			parseBegin := time.Now()
 			msg, err := parseMessageFromOneImporter(info, x.collector, impInfo)
 			if err != nil {
 				return err
 			}
+			parseNanos.Add(time.Since(parseBegin).Nanoseconds())
 			name := getExportedConfName(info, impInfo)
+			exportBegin := time.Now()
+			defer func() { exportNanos.Add(time.Since(exportBegin).Nanoseconds()) }()
 			if info.SheetOpts.Patch == tableaupb.Patch_PATCH_MERGE {
 				if info.ExtInfo.DryRun == options.DryRunPatch {
 					clonedMainMsg := proto.Clone(mainMsg)
@@ -101,7 +114,15 @@ func (x *sheetExporter) ScatterAndExport(info *SheetInfo,
 			return storeMessage(msg, name, info.LocationName, x.OutputDir, x.OutputOpt, x.validator)
 		})
 	}
-	return g.Wait()
+	if err := g.Wait(); err != nil {
+		return err
+	}
+	// NOTE: parse/export totals here are SUMS of per-job CPU-style durations
+	// across all scatter shards (not wall-clock). They reflect work volume,
+	// not elapsed time, since shards run concurrently.
+	log.Infof("%15s: %s confgen=%dms export=%dms (scatter, summed across shards)", "perf", info.MD.Name(),
+		parseNanos.Load()/int64(time.Millisecond), exportNanos.Load()/int64(time.Millisecond))
+	return nil
 }
 
 // MergeAndExport parses multiple importer infos and merges them into one
@@ -111,10 +132,12 @@ func (x *sheetExporter) MergeAndExport(info *SheetInfo,
 	impInfos ...importer.ImporterInfo) error {
 	// append main
 	allImpInfos := append(impInfos, importer.ImporterInfo{Importer: mainImpInfo})
+	parseBegin := time.Now()
 	protomsg, err := ParseMessage(info, x.collector, allImpInfos...)
 	if err != nil {
 		return err
 	}
+	parseElapsed := time.Since(parseBegin)
 	// exported conf name pattern is : [ParentDir/]<SheetName>
 	getExportedConfName := func(info *SheetInfo, impInfo importer.ImporterInfo) string {
 		// here filename has no ext suffix
@@ -126,7 +149,14 @@ func (x *sheetExporter) MergeAndExport(info *SheetInfo,
 		return filename
 	}
 	name := getExportedConfName(info, mainImpInfo)
-	return storeMessage(protomsg, name, info.LocationName, x.OutputDir, x.OutputOpt, x.validator)
+	exportBegin := time.Now()
+	if err := storeMessage(protomsg, name, info.LocationName, x.OutputDir, x.OutputOpt, x.validator); err != nil {
+		return err
+	}
+	exportElapsed := time.Since(exportBegin)
+	log.Infof("%15s: %s confgen=%dms export=%dms", "perf", info.MD.Name(),
+		parseElapsed.Milliseconds(), exportElapsed.Milliseconds())
+	return nil
 }
 
 type oneMsg struct {

internal/confgen/table_parser.go

@@ -11,6 +11,7 @@ import (
 	"github.com/tableauio/tableau/internal/types"
 	"github.com/tableauio/tableau/internal/x/xerrors"
 	"github.com/tableauio/tableau/internal/x/xproto"
+	"github.com/tableauio/tableau/log"
 	"github.com/tableauio/tableau/proto/tableaupb"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/reflect/protoreflect"
@@ -30,6 +31,18 @@ func (p *tableParser) Parse(protomsg proto.Message, sheet *book.Sheet) error {
 	} else {
 		table = sheet.Table
 	}
+	// Parallel confgen is enabled automatically whenever the sheet is eligible:
+	// callers no longer opt in via a worksheet/metasheet toggle. The
+	// eligibility predicate (canParallelizeSheet) is the sole authority --
+	// ineligible sheets transparently fall back to the serial path with a
+	// debug log explaining why, so a regression in eligibility never silently
+	// degrades correctness, only performance.
+	md := protomsg.ProtoReflect().Descriptor()
+	if plan := canParallelizeSheet(p.sheetParser, md); plan.ok {
+		return p.parseTableInParallel(protomsg, table, plan)
+	} else if plan.reason != "" {
+		log.Debugf("parallel confgen disabled for sheet %s: %s", p.sheetOpts.GetName(), plan.reason)
+	}
 	return p.parse(protomsg, table)
 }
 

internal/importer/book/tableparser/parser.go

@@ -11,10 +11,128 @@ func RangeDataRows(table book.Tabler, header *Header, fn func(*book.Row) error)
 	if err != nil {
 		return err
 	}
-	var prev *book.Row
-	// [datarow, endRow]: data rows
 	dataRow := table.BeginRow() + header.DataRow - 1
-	for row := dataRow; row < table.EndRow(); row++ {
+	return rangeDataRowsInRange(table, columns, lookupTable, header, dataRow, table.EndRow(), fn)
+}
+
+// RangeDataRowsInBlock ranges data rows in [blockBegin, blockEnd) of the table.
+//
+// Compared to RangeDataRows, the caller controls the row range, which is useful
+// for sharding the data rows across goroutines. The header (name/type rows) is
+// still parsed once from the table itself, so the column lookup table is
+// consistent across blocks.
+//
+// NOTE: blockBegin/blockEnd must lie within [dataRow, EndRow()) where dataRow
+// is `table.BeginRow() + header.DataRow - 1`. The previous-row chain (used for
+// adjacent_key auto-population) restarts at nil at the block boundary, so
+// callers MUST ensure the sheet does not rely on cross-block prev linkage
+// (e.g. AdjacentKey is not enabled).
+func RangeDataRowsInBlock(table book.Tabler, header *Header, blockBegin, blockEnd int, fn func(*book.Row) error) error {
+	columns, lookupTable, err := parseColumns(table, header)
+	if err != nil {
+		return err
+	}
+	return rangeDataRowsInRange(table, columns, lookupTable, header, blockBegin, blockEnd, fn)
+}
+
+// DataRowRange returns the [begin, end) data row range of the table according
+// to the header settings. It can be used by callers that need to shard the
+// data rows for parallel processing.
+func DataRowRange(table book.Tabler, header *Header) (int, int) {
+	return table.BeginRow() + header.DataRow - 1, table.EndRow()
+}
+
+// RangeDataRowsByIndices ranges data rows in the explicit `rowIndices` list of
+// the table, in the given order.
+//
+// Compared to RangeDataRowsInBlock (which iterates a contiguous half-open
+// interval), this variant lets callers pick any subset of data rows -- useful
+// for hash-partitioned parallel parsing where rows sharing the same outer key
+// must be processed by the same goroutine. Callers SHOULD pass `rowIndices` in
+// ascending order; the iteration faithfully follows the slice order, so any
+// non-ascending input would scramble within-shard row order and break the
+// invariant that "the merged result equals the serial result".
+//
+// The previous-row chain (used for adjacent_key auto-population) restarts at
+// nil at every call, so callers MUST ensure the sheet does not rely on
+// cross-row prev linkage (e.g. AdjacentKey is not enabled).
+func RangeDataRowsByIndices(table book.Tabler, header *Header, rowIndices []int, fn func(*book.Row) error) error {
+	columns, lookupTable, err := parseColumns(table, header)
+	if err != nil {
+		return err
+	}
+	return rangeDataRowsAtIndices(table, columns, lookupTable, header, rowIndices, fn)
+}
+
+func rangeDataRowsAtIndices(table book.Tabler, columns map[int]*book.Column, lookupTable book.ColumnLookupTable, header *Header, rowIndices []int, fn func(*book.Row) error) error {
+	var prev *book.Row
+	for _, row := range rowIndices {
+		curr := book.NewRow(row, prev, lookupTable)
+		for col := table.BeginCol(); col < table.EndCol(); col++ {
+			data, err := table.Cell(row, col)
+			if err != nil {
+				return xerrors.WrapKV(err)
+			}
+			curr.AddCell(columns[col], data, header.AdjacentKey)
+		}
+		ignored, err := curr.Ignored()
+		if err != nil {
+			return err
+		}
+		if ignored {
+			curr.Free()
+			continue
+		}
+		err = fn(curr)
+		if err != nil {
+			return err
+		}
+		if prev != nil {
+			prev.Free()
+		}
+		prev = curr
+	}
+	if prev != nil {
+		prev.Free()
+	}
+	return nil
+}
+
+// ScanColumn reads the raw cell strings of the named column for every data
+// row in [beginRow, endRow), returning a slice with one entry per row. It is
+// intended for cheap pre-scans that need to bucket rows by a single column
+// (e.g. parallel sharders that hash on the top-level outer-key column).
+//
+// The returned slice is parallel to [beginRow, endRow): index i corresponds
+// to row beginRow+i. ScanColumn does not honour `ignored` rows -- the caller
+// already accepts that the partition merely needs to be a function of (row,
+// key) and that any row-level filtering will happen later, inside the worker.
+//
+// If the named column is absent, ScanColumn returns an error so the caller
+// fails loudly rather than silently bucket all rows into a single shard.
+func ScanColumn(table book.Tabler, header *Header, name string, beginRow, endRow int) ([]string, error) {
+	_, lookupTable, err := parseColumns(table, header)
+	if err != nil {
+		return nil, err
+	}
+	col, ok := lookupTable[name]
+	if !ok {
+		return nil, xerrors.E2014(name)
+	}
+	values := make([]string, endRow-beginRow)
+	for row := beginRow; row < endRow; row++ {
+		data, cellErr := table.Cell(row, col)
+		if cellErr != nil {
+			return nil, xerrors.WrapKV(cellErr)
+		}
+		values[row-beginRow] = data
+	}
+	return values, nil
+}
+
+func rangeDataRowsInRange(table book.Tabler, columns map[int]*book.Column, lookupTable book.ColumnLookupTable, header *Header, beginRow, endRow int, fn func(*book.Row) error) error {
+	var prev *book.Row
+	for row := beginRow; row < endRow; row++ {
 		curr := book.NewRow(row, prev, lookupTable)
 		for col := table.BeginCol(); col < table.EndCol(); col++ {
 			data, err := table.Cell(row, col)

internal/x/xpool/sem.go

@@ -0,0 +1,101 @@
+// Package xpool provides lightweight concurrency-limiting primitives shared
+// across confgen / protogen pipelines.
+//
+// The package intentionally exposes a small surface:
+//
+//   - [Semaphore] is a counting semaphore implemented as a buffered channel.
+//     It is the building block for "at most N goroutines may execute the
+//     critical section concurrently" patterns. Acquisitions are
+//     context-cancellable so that errgroup-style fan-out propagates
+//     cancellation correctly.
+//
+// Why a custom type instead of golang.org/x/sync/semaphore? The standard
+// package supports weighted acquisitions and uses a sync.Mutex + waiter
+// list internally; we only ever acquire weight=1, and the chan-based
+// implementation is both simpler and faster on the hot path (single
+// channel send / receive, zero allocations after construction).
+package xpool
+
+import (
+	"context"
+	"runtime"
+)
+
+// Semaphore is a counting semaphore that bounds the number of goroutines
+// allowed to hold a "slot" simultaneously. It is safe for concurrent use
+// by multiple goroutines.
+//
+// The zero value is NOT usable; construct via [NewSemaphore] or
+// [NewCPUSemaphore].
+type Semaphore struct {
+	slots chan struct{}
+}
+
+// NewSemaphore constructs a semaphore with the given capacity. capacity
+// must be >= 1; values < 1 are clamped to 1 so that the semaphore always
+// admits at least one holder (the alternative -- a deadlocked semaphore --
+// would silently break callers that compute capacity from environment
+// signals like GOMAXPROCS in degenerate setups).
+func NewSemaphore(capacity int) *Semaphore {
+	if capacity < 1 {
+		capacity = 1
+	}
+	return &Semaphore{slots: make(chan struct{}, capacity)}
+}
+
+// NewCPUSemaphore constructs a semaphore sized to runtime.GOMAXPROCS(0),
+// the canonical "number of cores Go is allowed to use" signal that is
+// cgroup-aware on recent Go releases. This is the right size for CPU-
+// bound critical sections.
+func NewCPUSemaphore() *Semaphore {
+	return NewSemaphore(runtime.GOMAXPROCS(0))
+}
+
+// Cap reports the configured capacity.
+func (s *Semaphore) Cap() int { return cap(s.slots) }
+
+// Len reports the number of slots currently held. Intended for debugging
+// / observability; the value is approximate under concurrent mutation.
+func (s *Semaphore) Len() int { return len(s.slots) }
+
+// Acquire blocks until a slot is available or ctx is cancelled. On
+// success it returns nil and the caller MUST call [Semaphore.Release]
+// exactly once when done. On cancellation it returns ctx.Err() and the
+// caller MUST NOT call Release.
+func (s *Semaphore) Acquire(ctx context.Context) error {
+	select {
+	case s.slots <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// TryAcquire attempts to take a slot without blocking. It returns true
+// on success (in which case [Semaphore.Release] must be called) or
+// false if no slot was immediately available.
+func (s *Semaphore) TryAcquire() bool {
+	select {
+	case s.slots <- struct{}{}:
+		return true
+	default:
+		return false
+	}
+}
+
+// Release returns a previously acquired slot. It is the caller's
+// responsibility to ensure Release is called at most once per successful
+// Acquire / TryAcquire; a stray Release will silently consume a slot
+// from another holder, manifesting as a hard-to-diagnose hang.
+func (s *Semaphore) Release() { <-s.slots }
+
+// Run is a convenience helper that acquires a slot, runs fn, then
+// releases the slot. If ctx is cancelled before a slot becomes available
+// fn is NOT invoked and ctx.Err() is returned.
+func (s *Semaphore) Run(ctx context.Context, fn func() error) error {
+	if err := s.Acquire(ctx); err != nil {
+		return err
+	}
+	defer s.Release()
+	return fn()
+}