001package gudusoft.gsqlparser.parser;
002
003import gudusoft.gsqlparser.EDbVendor;
004import gudusoft.gsqlparser.ETokenType;
005import gudusoft.gsqlparser.TBaseType;
006import gudusoft.gsqlparser.TCustomLexer;
007import gudusoft.gsqlparser.TCustomParser;
008import gudusoft.gsqlparser.TCustomSqlStatement;
009import gudusoft.gsqlparser.TLexerFlink;
010import gudusoft.gsqlparser.TParserFlink;
011import gudusoft.gsqlparser.TSourceToken;
012import gudusoft.gsqlparser.TSourceTokenList;
013import gudusoft.gsqlparser.TStatementList;
014import gudusoft.gsqlparser.TSyntaxError;
015import gudusoft.gsqlparser.EFindSqlStateType;
016import gudusoft.gsqlparser.ETokenType;
017import gudusoft.gsqlparser.ETokenStatus;
018import gudusoft.gsqlparser.ESqlStatementType;
019import gudusoft.gsqlparser.EErrorType;
020import gudusoft.gsqlparser.stmt.TUnknownSqlStatement;
021import gudusoft.gsqlparser.stmt.mysql.TMySQLSource;
022import gudusoft.gsqlparser.sqlcmds.ISqlCmds;
023import gudusoft.gsqlparser.sqlcmds.SqlCmdsFactory;
024import gudusoft.gsqlparser.compiler.TContext;
025import gudusoft.gsqlparser.sqlenv.TSQLEnv;
026import gudusoft.gsqlparser.compiler.TGlobalScope;
027import gudusoft.gsqlparser.compiler.TFrame;
028import gudusoft.gsqlparser.resolver.TSQLResolver;
029import gudusoft.gsqlparser.TLog;
030import gudusoft.gsqlparser.compiler.TASTEvaluator;
031
032import java.io.BufferedReader;
033import java.util.ArrayList;
034import java.util.List;
035import java.util.Stack;
036
037/**
038 * Apache Flink SQL parser implementation.
039 *
040 * <p>This parser handles Flink SQL-specific syntax including:
041 * <ul>
042 *   <li>Flink SQL DML/DDL operations</li>
043 *   <li>Special token handling for DATE, TIME, TIMESTAMP, INTERVAL</li>
044 *   <li>Stored procedures, functions, and triggers</li>
045 *   <li>Flink-specific statements: CREATE CATALOG, CREATE MODEL, etc.</li>
046 * </ul>
047 *
048 * <p><b>Implementation Status:</b> NEW
049 * <ul>
050 *   <li><b>Base:</b> SparkSQL (Apache Calcite foundation)</li>
051 *   <li><b>Tokenization:</b> doflinktexttotokenlist()</li>
052 *   <li><b>Raw Extraction:</b> doflinkgetrawsqlstatements()</li>
053 *   <li><b>Parsing:</b> Fully self-contained using TParserFlink</li>
054 * </ul>
055 *
056 * @see SqlParser
057 * @see AbstractSqlParser
058 * @see TLexerFlink
059 * @see TParserFlink
060 * @since 3.2.0.0
061 */
062public class FlinkSqlParser extends AbstractSqlParser {
063
064    // Vendor-specific parser and lexer
065    private TLexerFlink flexer;
066    private TParserFlink fparser;
067
068    // State management for raw statement extraction
069    private TCustomSqlStatement gcurrentsqlstatement;
070    private String userDelimiterStr;
071    private char curdelimiterchar;
072
073    /**
074     * Construct Flink SQL parser.
075     * <p>
076     * Configures the parser for Flink SQL with default delimiter (;).
077     */
078    public FlinkSqlParser() {
079        super(EDbVendor.dbvflink);
080        this.delimiterChar = ';';  // Flink SQL delimiter
081        this.defaultDelimiterStr = ";";  // Default delimiter
082
083        // Create lexer once - will be reused for all parsing operations
084        this.flexer = new TLexerFlink();
085        this.flexer.delimiterchar = this.delimiterChar;
086        this.flexer.defaultDelimiterStr = this.defaultDelimiterStr;
087
088        // Set parent's lexer reference for shared tokenization logic
089        this.lexer = this.flexer;
090
091        // Create parser once - will be reused for all parsing operations
092        this.fparser = new TParserFlink(null);
093        this.fparser.lexer = this.flexer;
094    }
095
096    @Override
097    public EDbVendor getVendor() {
098        return vendor;
099    }
100
101    // ========== Abstract Method Implementations ==========
102
103    @Override
104    protected TCustomLexer getLexer(ParserContext context) {
105        return this.flexer;
106    }
107
108    @Override
109    protected TCustomParser getParser(ParserContext context, TSourceTokenList tokens) {
110        return this.fparser;
111    }
112
113    @Override
114    protected TCustomParser getSecondaryParser(ParserContext context, TSourceTokenList tokens) {
115        // Flink SQL doesn't have a secondary parser
116        return null;
117    }
118
119    // ========== Tokenization Phase (Hook Pattern) ==========
120
121    /**
122     * Hook method for vendor-specific tokenization.
123     * <p>
124     * Delegates to doflinktexttotokenlist() which implements Flink SQL-specific
125     * token processing logic.
126     */
127    @Override
128    protected void tokenizeVendorSql() {
129        doflinktexttotokenlist();
130    }
131
132    /**
133     * Flink SQL-specific tokenization logic.
134     * <p>
135     * Special handling:
136     * <ul>
137     *   <li>MySQL-style comment validation</li>
138     *   <li>WITH ROLLUP token adjustment</li>
139     *   <li>Delimiter detection</li>
140     * </ul>
141     */
142    private void doflinktexttotokenlist() {
143        TSourceToken asourcetoken, lcprevst;
144        int yychar;
145        boolean startDelimiter = false;
146
147        flexer.tmpDelimiter = "";
148
149        asourcetoken = getanewsourcetoken();
150        if (asourcetoken == null) return;
151        yychar = asourcetoken.tokencode;
152
153        while (yychar > 0) {
154            sourcetokenlist.add(asourcetoken);
155            asourcetoken = getanewsourcetoken();
156            if (asourcetoken == null) break;
157            checkFlinkCommentToken(asourcetoken);
158
159            if ((asourcetoken.tokencode == TBaseType.lexnewline) && (startDelimiter)) {
160                startDelimiter = false;
161                flexer.tmpDelimiter = sourcetokenlist.get(sourcetokenlist.size() - 1).getAstext();
162            }
163
164            if (asourcetoken.tokencode == TBaseType.rrw_rollup) {
165                // with rollup
166                lcprevst = getprevsolidtoken(asourcetoken);
167                if (lcprevst != null) {
168                    if (lcprevst.tokencode == TBaseType.rrw_with)
169                        lcprevst.tokencode = TBaseType.with_rollup;
170                }
171            }
172
173            yychar = asourcetoken.tokencode;
174        }
175    }
176
177    /**
178     * Helper method for Flink-style comment validation.
179     */
180    private void checkFlinkCommentToken(TSourceToken cmtToken) {
181        // No-op: similar to SparkSQL
182    }
183
184    /**
185     * Helper method to get previous solid token (non-whitespace, non-comment).
186     */
187    private TSourceToken getprevsolidtoken(TSourceToken ptoken) {
188        TSourceToken lcprevtoken = null;
189        int i = ptoken.posinlist;
190        while (i > 0) {
191            i--;
192            lcprevtoken = sourcetokenlist.get(i);
193            if ((lcprevtoken.tokencode == TBaseType.lexspace)
194                    || (lcprevtoken.tokencode == TBaseType.lexnewline)
195                    || (lcprevtoken.tokencode == TBaseType.cmtdoublehyphen)
196                    || (lcprevtoken.tokencode == TBaseType.cmtslashstar)) {
197                continue;
198            }
199            return lcprevtoken;
200        }
201        return null;
202    }
203
204    /**
205     * Helper method to add token to statement.
206     */
207    private void appendToken(TCustomSqlStatement statement, TSourceToken token) {
208        if (statement == null || token == null) {
209            return;
210        }
211        token.stmt = statement;
212        statement.sourcetokenlist.add(token);
213    }
214
215    // ========== Raw Statement Extraction Phase (Hook Pattern) ==========
216
217    /**
218     * Hook method to setup parsers before raw statement extraction.
219     */
220    @Override
221    protected void setupVendorParsersForExtraction() {
222        this.fparser.sqlcmds = this.sqlcmds;
223        this.fparser.sourcetokenlist = this.sourcetokenlist;
224    }
225
226    /**
227     * Hook method for vendor-specific raw statement extraction.
228     */
229    @Override
230    protected void extractVendorRawStatements(SqlParseResult.Builder builder) {
231        doflinkgetrawsqlstatements(builder);
232    }
233
234    /**
235     * Flink SQL-specific raw statement extraction logic.
236     * <p>
237     * This method:
238     * <ul>
239     *   <li>Adjusts DATE, TIME, TIMESTAMP, INTERVAL token codes based on context</li>
240     *   <li>Handles statement boundaries (semicolon, custom delimiters)</li>
241     *   <li>Supports stored procedures with BEGIN/END blocks</li>
242     * </ul>
243     *
244     * @param builder the result builder to collect errors
245     * @return error count (currently always 0)
246     */
247    private int doflinkgetrawsqlstatements(SqlParseResult.Builder builder) {
248        int errorcount = 0;
249        gcurrentsqlstatement = null;
250        EFindSqlStateType gst = EFindSqlStateType.stnormal;
251        int i;
252        TSourceToken ast;
253        boolean waitingDelimiter = false;
254
255        // Reset delimiter
256        userDelimiterStr = defaultDelimiterStr;
257
258        for (i = 0; i < sourcetokenlist.size(); i++) {
259            ast = sourcetokenlist.get(i);
260            sourcetokenlist.curpos = i;
261
262            // Flink SQL-specific token adjustments (similar to SparkSQL)
263            if (ast.tokencode == TBaseType.rrw_date) {
264                TSourceToken st1 = ast.nextSolidToken();
265                if (st1 != null) {
266                    if (st1.tokencode == '(') {
267                        ast.tokencode = TBaseType.rrw_spark_date_function;
268                    } else if (st1.tokencode == TBaseType.sconst) {
269                        ast.tokencode = TBaseType.rrw_spark_date_const;
270                    }
271                }
272            } else if (ast.tokencode == TBaseType.rrw_time) {
273                TSourceToken st1 = ast.nextSolidToken();
274                if (st1 != null) {
275                    if (st1.tokencode == TBaseType.sconst) {
276                        ast.tokencode = TBaseType.rrw_spark_time_const;
277                    }
278                }
279            } else if (ast.tokencode == TBaseType.rrw_timestamp) {
280                TSourceToken st1 = ast.nextSolidToken();
281                if (st1 != null) {
282                    if (st1.tokencode == TBaseType.sconst) {
283                        ast.tokencode = TBaseType.rrw_spark_timestamp_constant;
284                    } else if (st1.tokencode == TBaseType.ident) {
285                        if (st1.toString().startsWith("\"")) {
286                            ast.tokencode = TBaseType.rrw_spark_timestamp_constant;
287                            st1.tokencode = TBaseType.sconst;
288                        }
289                    }
290                }
291            } else if (ast.tokencode == TBaseType.rrw_interval) {
292                TSourceToken leftParen = ast.searchToken('(', 1);
293                if (leftParen != null) {
294                    int k = leftParen.posinlist + 1;
295                    boolean commaToken = false;
296                    while (k < ast.container.size()) {
297                        if (ast.container.get(k).tokencode == ')') break;
298                        if (ast.container.get(k).tokencode == ',') {
299                            commaToken = true;
300                            break;
301                        }
302                        k++;
303                    }
304                    if (commaToken) {
305                        ast.tokencode = TBaseType.rrw_mysql_interval_func;
306                    }
307                }
308            } else if (ast.tokencode == TBaseType.rrw_spark_position) {
309                TSourceToken leftParen = ast.searchToken('(', 1);
310                if (leftParen != null) {
311                    // POSITION is a function
312                } else {
313                    ast.tokencode = TBaseType.ident; // treat it as identifier
314                }
315            }
316            // Handle LOCALTIME: distinguish function from identifier based on context
317            // Function context: SELECT LOCALTIME, WHERE LOCALTIME > x
318            // Identifier context: SELECT t.localtime (after period - field reference)
319            // Note: Use flink_rw_localtime (707) which matches the Flink lexer's token code
320            else if (ast.tokencode == TBaseType.flink_rw_localtime) {
321                TSourceToken prevToken = ast.prevSolidToken();
322                // If preceded by period, it's a field reference (identifier)
323                // Otherwise, it's a builtin function
324                if (prevToken == null || prevToken.tokencode != '.') {
325                    ast.tokencode = TBaseType.rrw_flink_localtime_as_func;
326                }
327            }
328            // Handle LOCALTIMESTAMP: same logic as LOCALTIME
329            // Note: Use flink_rw_localtimestamp (708) which matches the Flink lexer's token code
330            else if (ast.tokencode == TBaseType.flink_rw_localtimestamp) {
331                TSourceToken prevToken = ast.prevSolidToken();
332                if (prevToken == null || prevToken.tokencode != '.') {
333                    ast.tokencode = TBaseType.rrw_flink_localtimestamp_as_func;
334                }
335            }
336            // Named argument assignment "=>" in window TVF calls: TUMBLE(DATA => TABLE t, ...).
337            // The lexer returns a generic multi-char operator; map it to RW_FLINK_NAMED_ARG.
338            else if ((ast.tokentype == ETokenType.ttmulticharoperator) && ("=>".equals(ast.toString()))) {
339                ast.tokencode = TBaseType.rrw_flink_named_arg;
340            }
341
342            switch (gst) {
343                case sterror: {
344                    if (ast.tokentype == ETokenType.ttsemicolon) {
345                        appendToken(gcurrentsqlstatement, ast);
346                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
347                        gst = EFindSqlStateType.stnormal;
348                    } else {
349                        appendToken(gcurrentsqlstatement, ast);
350                    }
351                    break;
352                }
353                case stnormal: {
354                    if ((ast.tokencode == TBaseType.cmtdoublehyphen)
355                            || (ast.tokencode == TBaseType.cmtslashstar)
356                            || (ast.tokencode == TBaseType.lexspace)
357                            || (ast.tokencode == TBaseType.lexnewline)
358                            || (ast.tokentype == ETokenType.ttsemicolon)) {
359                        if (TBaseType.assigned(gcurrentsqlstatement)) {
360                            appendToken(gcurrentsqlstatement, ast);
361                        }
362                        continue;
363                    }
364
365                    if ((ast.isFirstTokenOfLine()) && ((ast.tokencode == TBaseType.rrw_mysql_source) || (ast.tokencode == TBaseType.slash_dot))) {
366                        gst = EFindSqlStateType.stsqlplus;
367                        gcurrentsqlstatement = new TMySQLSource(vendor);
368                        appendToken(gcurrentsqlstatement, ast);
369                        continue;
370                    }
371
372                    // Find a token to start sql or plsql mode
373                    gcurrentsqlstatement = sqlcmds.issql(ast, gst, gcurrentsqlstatement);
374
375                    if (TBaseType.assigned(gcurrentsqlstatement)) {
376                        ESqlStatementType[] ses = {ESqlStatementType.sstmysqlcreateprocedure, ESqlStatementType.sstmysqlcreatefunction,
377                                ESqlStatementType.sstcreateprocedure, ESqlStatementType.sstcreatefunction,
378                                ESqlStatementType.sstcreatetrigger};
379                        if (includesqlstatementtype(gcurrentsqlstatement.sqlstatementtype, ses)) {
380                            gst = EFindSqlStateType.ststoredprocedure;
381                            waitingDelimiter = false;
382                            appendToken(gcurrentsqlstatement, ast);
383                            curdelimiterchar = ';';
384                        } else {
385                            gst = EFindSqlStateType.stsql;
386                            appendToken(gcurrentsqlstatement, ast);
387                        }
388                    }
389
390                    if (!TBaseType.assigned(gcurrentsqlstatement)) {
391                        // Error token found
392                        this.syntaxErrors.add(new TSyntaxError(ast.getAstext(), ast.lineNo, (ast.columnNo < 0 ? 0 : ast.columnNo),
393                                "Error when tokenlize", EErrorType.spwarning, TBaseType.MSG_WARNING_ERROR_WHEN_TOKENIZE, null, ast.posinlist));
394
395                        ast.tokentype = ETokenType.tttokenlizererrortoken;
396                        gst = EFindSqlStateType.sterror;
397
398                        gcurrentsqlstatement = new TUnknownSqlStatement(vendor);
399                        gcurrentsqlstatement.sqlstatementtype = ESqlStatementType.sstinvalid;
400                        appendToken(gcurrentsqlstatement, ast);
401                    }
402                    break;
403                }
404                case stsqlplus: {
405                    if (ast.tokencode == TBaseType.lexnewline) {
406                        gst = EFindSqlStateType.stnormal;
407                        appendToken(gcurrentsqlstatement, ast);
408                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
409                    } else {
410                        appendToken(gcurrentsqlstatement, ast);
411                    }
412                    break;
413                }
414                case stsql: {
415                    if ((ast.tokentype == ETokenType.ttsemicolon) && (gcurrentsqlstatement.sqlstatementtype != ESqlStatementType.sstmysqldelimiter)) {
416                        gst = EFindSqlStateType.stnormal;
417                        appendToken(gcurrentsqlstatement, ast);
418                        gcurrentsqlstatement.semicolonended = ast;
419                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
420                        continue;
421                    }
422                    if (ast.toString().equalsIgnoreCase(userDelimiterStr)) {
423                        gst = EFindSqlStateType.stnormal;
424                        ast.tokencode = ';'; // treat it as semicolon
425                        appendToken(gcurrentsqlstatement, ast);
426                        gcurrentsqlstatement.semicolonended = ast;
427                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
428                        continue;
429                    }
430                    appendToken(gcurrentsqlstatement, ast);
431
432                    if ((ast.tokencode == TBaseType.lexnewline)
433                            && (gcurrentsqlstatement.sqlstatementtype == ESqlStatementType.sstmysqldelimiter)) {
434                        gst = EFindSqlStateType.stnormal;
435                        userDelimiterStr = "";
436                        for (int k = 0; k < gcurrentsqlstatement.sourcetokenlist.size(); k++) {
437                            TSourceToken st = gcurrentsqlstatement.sourcetokenlist.get(k);
438                            if ((st.tokencode == TBaseType.rrw_mysql_delimiter)
439                                    || (st.tokencode == TBaseType.lexnewline)
440                                    || (st.tokencode == TBaseType.lexspace)
441                                    || (st.tokencode == TBaseType.rrw_set)) {
442                                continue;
443                            }
444                            userDelimiterStr += st.toString();
445                        }
446                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
447                        continue;
448                    }
449                    break;
450                }
451                case ststoredprocedure: {
452                    // Single stmt in function/procedure/trigger may use ; as terminate char
453                    if (waitingDelimiter) {
454                        if (userDelimiterStr.equalsIgnoreCase(ast.toString())) {
455                            gst = EFindSqlStateType.stnormal;
456                            gcurrentsqlstatement.semicolonended = ast;
457                            onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
458                            continue;
459                        } else if (userDelimiterStr.startsWith(ast.toString())) {
460                            String lcstr = ast.toString();
461                            for (int k = ast.posinlist + 1; k < ast.container.size(); k++) {
462                                TSourceToken st = ast.container.get(k);
463                                if ((st.tokencode == TBaseType.rrw_mysql_delimiter) || (st.tokencode == TBaseType.lexnewline) || (st.tokencode == TBaseType.lexspace)) {
464                                    break;
465                                }
466                                lcstr = lcstr + st.toString();
467                            }
468
469                            if (userDelimiterStr.equalsIgnoreCase(lcstr)) {
470                                for (int k = ast.posinlist; k < ast.container.size(); k++) {
471                                    TSourceToken st = ast.container.get(k);
472                                    if ((st.tokencode == TBaseType.rrw_mysql_delimiter) || (st.tokencode == TBaseType.lexnewline) || (st.tokencode == TBaseType.lexspace)) {
473                                        break;
474                                    }
475                                    ast.tokenstatus = ETokenStatus.tsignorebyyacc;
476                                }
477                                gst = EFindSqlStateType.stnormal;
478                                gcurrentsqlstatement.semicolonended = ast;
479                                onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
480                                continue;
481                            }
482                        }
483                    }
484                    if (ast.tokencode == TBaseType.rrw_begin)
485                        waitingDelimiter = true;
486
487                    if (userDelimiterStr.equals(";") || (waitingDelimiter == false)) {
488                        appendToken(gcurrentsqlstatement, ast);
489                        if (ast.tokentype == ETokenType.ttsemicolon) {
490                            gst = EFindSqlStateType.stnormal;
491                            gcurrentsqlstatement.semicolonended = ast;
492                            onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
493                            continue;
494                        }
495                    } else {
496                        if (ast.toString().equals(userDelimiterStr)) {
497                            ast.tokenstatus = ETokenStatus.tsignorebyyacc;
498                            appendToken(gcurrentsqlstatement, ast);
499                            gst = EFindSqlStateType.stnormal;
500                            onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
501                        } else {
502                            if ((ast.tokentype == ETokenType.ttsemicolon) && (userDelimiterStr.equals(";"))) {
503                                TSourceToken lcprevtoken = ast.container.nextsolidtoken(ast, -1, false);
504                                if (lcprevtoken != null) {
505                                    if (lcprevtoken.tokencode == TBaseType.rrw_end) {
506                                        gst = EFindSqlStateType.stnormal;
507                                        gcurrentsqlstatement.semicolonended = ast;
508                                        appendToken(gcurrentsqlstatement, ast);
509                                        onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, false, builder);
510                                        continue;
511                                    }
512                                }
513                            }
514                            appendToken(gcurrentsqlstatement, ast);
515                        }
516                    }
517                    break;
518                }
519            }
520        }
521
522        // Last statement
523        if (TBaseType.assigned(gcurrentsqlstatement) && ((gst == EFindSqlStateType.stsql) || (gst == EFindSqlStateType.ststoredprocedure) || (gst == EFindSqlStateType.sterror))) {
524            onRawStatementComplete(this.parserContext, gcurrentsqlstatement, this.fparser, null, this.sqlstatements, true, builder);
525        }
526
527        // Populate builder with results
528        builder.sqlStatements(this.sqlstatements);
529        builder.syntaxErrors(syntaxErrors instanceof ArrayList ?
530                (ArrayList<TSyntaxError>) syntaxErrors : new ArrayList<>(syntaxErrors));
531        builder.errorCode(syntaxErrors.isEmpty() ? 0 : syntaxErrors.size());
532        builder.errorMessage(syntaxErrors.isEmpty() ? "" :
533                String.format("Raw extraction completed with %d error(s)", syntaxErrors.size()));
534
535        return errorcount;
536    }
537
538    /**
539     * Helper method to check if statement type is in the given array.
540     */
541    private boolean includesqlstatementtype(ESqlStatementType type, ESqlStatementType[] types) {
542        for (ESqlStatementType t : types) {
543            if (type == t) return true;
544        }
545        return false;
546    }
547
548    // ========== Parsing Phase ==========
549
550    @Override
551    protected TStatementList performParsing(ParserContext context, TCustomParser mainParser,
552                                            TCustomParser secondaryParser, TSourceTokenList tokens,
553                                            TStatementList rawStatements) {
554        // Store references
555        this.parserContext = context;
556        this.sourcetokenlist = tokens;
557        this.sqlstatements = rawStatements;
558
559        // Initialize sqlcmds for this vendor
560        this.sqlcmds = SqlCmdsFactory.get(vendor);
561
562        // Inject sqlcmds into parser
563        this.fparser.sqlcmds = this.sqlcmds;
564
565        // Initialize global context
566        initializeGlobalContext();
567
568        // Parse each statement
569        for (int i = 0; i < sqlstatements.size(); i++) {
570            TCustomSqlStatement stmt = sqlstatements.getRawSql(i);
571            try {
572                stmt.setFrameStack(frameStack);
573                int parseResult = stmt.parsestatement(null, false, context.isOnlyNeedRawParseTree());
574
575                // Vendor-specific post-processing
576                afterStatementParsed(stmt);
577
578                // Error recovery
579                boolean doRecover = TBaseType.ENABLE_ERROR_RECOVER_IN_CREATE_TABLE;
580                if (doRecover && ((parseResult != 0) || (stmt.getErrorCount() > 0))) {
581                    handleCreateTableErrorRecovery(stmt);
582                }
583
584                // Collect errors
585                if ((parseResult != 0) || (stmt.getErrorCount() > 0)) {
586                    copyErrorsFromStatement(stmt);
587                }
588            } catch (Exception ex) {
589                handleStatementParsingException(stmt, i, ex);
590                continue;
591            }
592        }
593
594        // Clean up frame stack
595        if (globalFrame != null) globalFrame.popMeFromStack(frameStack);
596
597        return sqlstatements;
598    }
599
600    /**
601     * Hook for vendor-specific post-processing after statement is parsed.
602     */
603    protected void afterStatementParsed(TCustomSqlStatement stmt) {
604        // No special post-processing needed for Flink SQL
605    }
606
607    /**
608     * Handle error recovery for CREATE TABLE statements.
609     */
610    protected void handleCreateTableErrorRecovery(TCustomSqlStatement stmt) {
611        if (((stmt.sqlstatementtype == ESqlStatementType.sstcreatetable) ||
612             ((stmt.sqlstatementtype == ESqlStatementType.sstcreateindex) && (vendor != EDbVendor.dbvcouchbase))) &&
613            (!TBaseType.c_createTableStrictParsing)) {
614
615            int nested = 0;
616            boolean isIgnore = false, isFoundIgnoreToken = false;
617            TSourceToken firstIgnoreToken = null;
618
619            for (int k = 0; k < stmt.sourcetokenlist.size(); k++) {
620                TSourceToken st = stmt.sourcetokenlist.get(k);
621
622                if (isIgnore) {
623                    if (st.issolidtoken() && (st.tokencode != ';')) {
624                        isFoundIgnoreToken = true;
625                        if (firstIgnoreToken == null) {
626                            firstIgnoreToken = st;
627                        }
628                    }
629                    if (st.tokencode != ';') {
630                        st.tokencode = TBaseType.sqlpluscmd;
631                    }
632                    continue;
633                }
634
635                if (st.tokencode == (int) ')') {
636                    nested--;
637                    if (nested == 0) {
638                        boolean isSelect = false;
639                        TSourceToken st1 = st.searchToken(TBaseType.rrw_as, 1);
640                        if (st1 != null) {
641                            TSourceToken st2 = st.searchToken((int) '(', 2);
642                            if (st2 != null) {
643                                TSourceToken st3 = st.searchToken(TBaseType.rrw_select, 3);
644                                isSelect = (st3 != null);
645                            }
646                        }
647                        if (!isSelect) isIgnore = true;
648                    }
649                }
650
651                if ((st.tokencode == (int) '(') || (st.tokencode == TBaseType.left_parenthesis_2)) {
652                    nested++;
653                }
654            }
655
656            // Retry parsing if we found ignoreable properties
657            if (isFoundIgnoreToken) {
658                stmt.clearError();
659                stmt.parsestatement(null, false, this.parserContext.isOnlyNeedRawParseTree());
660            }
661        }
662    }
663
664    // ========== Semantic Analysis ==========
665
666    @Override
667    protected void performSemanticAnalysis(ParserContext context, TStatementList statements) {
668        if (!TBaseType.isEnableResolver()) {
669            return;
670        }
671
672        if (!getSyntaxErrors().isEmpty()) {
673            return;
674        }
675
676        // Run semantic resolver
677        TSQLResolver resolver = new TSQLResolver(globalContext, statements);
678        resolver.resolve();
679    }
680
681    // ========== Interpretation ==========
682
683    @Override
684    protected void performInterpreter(ParserContext context, TStatementList statements) {
685        if (!TBaseType.ENABLE_INTERPRETER) {
686            return;
687        }
688        // Flink SQL interpretation not currently supported
689    }
690
691    @Override
692    public String toString() {
693        return "FlinkSqlParser{vendor=" + vendor + "}";
694    }
695}