stream

Author: mstgnz

mysql/mysql_stream.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"sync"
 
 	"github.com/mstgnz/sqlmapper"
 )
@@ -127,6 +128,124 @@ func (p *MySQLStreamParser) ParseStream(reader io.Reader, callback func(sqlmappe
 	return nil
 }
 
+// ParseStreamParallel implements parallel processing for MySQL stream parsing
+func (p *MySQLStreamParser) ParseStreamParallel(reader io.Reader, callback func(sqlmapper.SchemaObject) error, workers int) error {
+	streamReader := sqlmapper.NewStreamReader(reader, ";")
+	statements := make(chan string, workers)
+	results := make(chan sqlmapper.SchemaObject, workers)
+	errors := make(chan error, workers)
+	var wg sync.WaitGroup
+
+	// Start worker goroutines
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for statement := range statements {
+				obj, err := p.parseStatement(statement)
+				if err != nil {
+					errors <- err
+					return
+				}
+				if obj != nil {
+					results <- *obj
+				}
+			}
+		}()
+	}
+
+	// Start a goroutine to close results channel after all workers are done
+	go func() {
+		wg.Wait()
+		close(results)
+	}()
+
+	// Start a goroutine to read statements and send them to workers
+	go func() {
+		for {
+			statement, err := streamReader.ReadStatement()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				errors <- fmt.Errorf("error reading statement: %v", err)
+				break
+			}
+
+			statement = strings.TrimSpace(statement)
+			if statement == "" {
+				continue
+			}
+			statements <- statement
+		}
+		close(statements)
+	}()
+
+	// Process results and handle errors
+	for obj := range results {
+		if err := callback(obj); err != nil {
+			return err
+		}
+	}
+
+	// Check for any errors from workers
+	select {
+	case err := <-errors:
+		return err
+	default:
+		return nil
+	}
+}
+
+// parseStatement parses a single SQL statement and returns a SchemaObject
+func (p *MySQLStreamParser) parseStatement(statement string) (*sqlmapper.SchemaObject, error) {
+	upperStatement := strings.ToUpper(statement)
+
+	switch {
+	case strings.HasPrefix(upperStatement, "CREATE TABLE"):
+		table, err := p.parseTableStatement(statement)
+		if err != nil {
+			return nil, err
+		}
+		return &sqlmapper.SchemaObject{
+			Type: sqlmapper.TableObject,
+			Data: table,
+		}, nil
+
+	case strings.HasPrefix(upperStatement, "CREATE VIEW"):
+		view, err := p.parseViewStatement(statement)
+		if err != nil {
+			return nil, err
+		}
+		return &sqlmapper.SchemaObject{
+			Type: sqlmapper.ViewObject,
+			Data: view,
+		}, nil
+
+	case strings.HasPrefix(upperStatement, "CREATE FUNCTION"):
+		function, err := p.parseFunctionStatement(statement)
+		if err != nil {
+			return nil, err
+		}
+		return &sqlmapper.SchemaObject{
+			Type: sqlmapper.FunctionObject,
+			Data: function,
+		}, nil
+
+	case strings.HasPrefix(upperStatement, "CREATE PROCEDURE"):
+		procedure, err := p.parseProcedureStatement(statement)
+		if err != nil {
+			return nil, err
+		}
+		return &sqlmapper.SchemaObject{
+			Type: sqlmapper.ProcedureObject,
+			Data: procedure,
+		}, nil
+	}
+
+	return nil, nil
+}
+
 // GenerateStream implements the StreamParser interface
 func (p *MySQLStreamParser) GenerateStream(schema *sqlmapper.Schema, writer io.Writer) error {
 	// Generate tables

oracle/oracle.go

@@ -556,3 +556,38 @@ func (o *Oracle) Generate(schema *sqlmapper.Schema) (string, error) {
 
 	return result.String(), nil
 }
+
+func (o *Oracle) parseTables(statement string) error {
+	// TODO: Implement table parsing
+	return nil
+}
+
+func (o *Oracle) parseViews(statement string) error {
+	// TODO: Implement view parsing
+	return nil
+}
+
+func (o *Oracle) parseFunctions(statement string) error {
+	// TODO: Implement function parsing
+	return nil
+}
+
+func (o *Oracle) parseTriggers(statement string) error {
+	// TODO: Implement trigger parsing
+	return nil
+}
+
+func (o *Oracle) parseSequences(statement string) error {
+	// TODO: Implement sequence parsing
+	return nil
+}
+
+func (o *Oracle) parseTypes(statement string) error {
+	// TODO: Implement type parsing
+	return nil
+}
+
+func (o *Oracle) parseIndexes(statement string) error {
+	// TODO: Implement index parsing
+	return nil
+}