Source code

001package gudusoft.gsqlparser.util;
002
003import gudusoft.gsqlparser.EDbVendor;
004import gudusoft.gsqlparser.TGSqlParser;
005import gudusoft.gsqlparser.util.json.JSON;
006
007import java.io.*;
008import java.util.ArrayList;
009import java.util.LinkedHashMap;
010import java.util.List;
011import java.util.Map;
012
013public class FileSplitter {
014
015    private static final int MAX_CONSECUTIVE_FAILURES = 3;
016    private static final int BUFFER_SIZE = 8192;
017    private static final Logger logger = LoggerFactory.getLogger(FileSplitter.class);
018
019    /**
020     * Split large file
021     *
022     * @param inputFile   Input file
023     * @param outputDir   Output directory
024     * @param splitSizeMB Split size (MB)
025     * @param dbVendor    Database vendor
026     * @return List of split files
027     * @throws IOException IO exception
028     */
029    public static List<File> splitFile(File inputFile, File outputDir, int splitSizeMB, EDbVendor dbVendor) throws IOException {
030        List<File> splitFiles = new ArrayList<>();
031        if (!inputFile.exists()) {
032            throw new FileNotFoundException("Input file does not exist: " + inputFile.getAbsolutePath());
033        }
034
035        if (!outputDir.exists() && !outputDir.mkdirs()) {
036            throw new IOException("Failed to create output directory: " + outputDir.getAbsolutePath());
037        }
038
039        // Check if it's a JSON file
040        if (isJsonFile(inputFile)) {
041            return splitJsonFile(inputFile, outputDir, splitSizeMB);
042        }
043
044        // Process regular SQL file
045        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
046
047        try (RandomAccessFile raf = new RandomAccessFile(inputFile, "r")) {
048            long fileLength = raf.length();
049            long currentPos = 0;
050            long chunkStartPos = 0;
051            int consecutiveFailures = 0;
052            int fileIndex = 1;
053            long globalStartLineNo = 1;
054
055            while (currentPos < fileLength) {
056                Long lastValidatedLineNo = getLastLineFromCurrentPos(raf, currentPos, fileLength, splitSizeBytes, dbVendor);
057                if (lastValidatedLineNo == null) break;
058
059                while (lastValidatedLineNo != null && lastValidatedLineNo == -1) {
060                    consecutiveFailures++;
061                    logger.warn("No validated statement found at position " + currentPos + "  (consecutive failures: " + consecutiveFailures + ")");
062
063                    long incrementSize = Math.min(splitSizeBytes, fileLength - currentPos);
064                    long previousPos = currentPos;
065                    currentPos = Math.min(currentPos + incrementSize, fileLength);
066
067                    if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
068                        logger.error("Reached maximum consecutive failures (" + MAX_CONSECUTIVE_FAILURES + "), skipping this segment and moving to next");
069                        long skipSize = splitSizeBytes * (consecutiveFailures + 1);
070                        currentPos = Math.min(previousPos + skipSize, fileLength);
071                        chunkStartPos = currentPos;
072                        consecutiveFailures = 0; // Reset consecutive failure counter
073                        logger.info("Skipped to position " + currentPos + ", continuing with next segment");
074                        break;
075                    }
076
077                    lastValidatedLineNo = getLastLineFromCurrentPos(raf, currentPos, fileLength, splitSizeBytes * (consecutiveFailures + 1), dbVendor);
078                }
079
080                if (lastValidatedLineNo == null) {
081                    break;
082                }
083
084                if (lastValidatedLineNo == -1) {
085                    continue;
086                }
087
088                consecutiveFailures = 0;
089                long endPos = findLineEndPosition(raf, currentPos, lastValidatedLineNo);
090                long globalEndLineNo = globalStartLineNo + lastValidatedLineNo - 1;
091
092                File splitFile = splitFileByPosition(outputDir, inputFile, chunkStartPos, endPos, fileIndex, globalStartLineNo, globalEndLineNo);
093                splitFiles.add(splitFile);
094
095                chunkStartPos = endPos;
096                currentPos = endPos;
097                globalStartLineNo = globalEndLineNo + 1;
098                fileIndex++;
099
100                if (currentPos >= fileLength) {
101                    break;
102                }
103            }
104        }
105
106        return splitFiles;
107    }
108
109    /**
110     * Check if file is a JSON file
111     */
112    private static boolean isJsonFile(File file) {
113        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
114            int c;
115            // Skip whitespace characters
116            while ((c = reader.read()) != -1) {
117                if (!Character.isWhitespace(c)) {
118                    // Check if first non-whitespace character is '{'
119                    return c == '{';
120                }
121            }
122        } catch (IOException e) {
123            logger.error("Error reading file to check if it's JSON: " + file.getAbsolutePath(), e);
124        }
125        return false;
126    }
127
128    /**
129     * Get file extension
130     */
131    private static String getFileExtension(String fileName) {
132        int lastDotIndex = fileName.lastIndexOf('.');
133        if (lastDotIndex == -1) {
134            return "";
135        }
136        return fileName.substring(lastDotIndex + 1);
137    }
138
139    /**
140     * Get file name without extension
141     */
142    private static String getFileNameWithoutExtension(String fileName) {
143        int lastDotIndex = fileName.lastIndexOf('.');
144        if (lastDotIndex == -1) {
145            return fileName;
146        }
147        return fileName.substring(0, lastDotIndex);
148    }
149
150    /**
151     * Move file
152     */
153    private static void moveFile(File source, File target) throws IOException {
154        if (!source.exists()) {
155            throw new FileNotFoundException("Source file does not exist: " + source.getAbsolutePath());
156        }
157        if (target.exists() && !target.delete()) {
158            throw new IOException("Failed to delete existing target file: " + target.getAbsolutePath());
159        }
160        if (!source.renameTo(target)) {
161            // If renameTo fails, try copy and delete
162            copyFile(source, target);
163            if (!source.delete()) {
164                throw new IOException("Failed to delete source file after copy: " + source.getAbsolutePath());
165            }
166        }
167    }
168
169    /**
170     * Copy file
171     */
172    private static void copyFile(File source, File target) throws IOException {
173        try (FileInputStream fis = new FileInputStream(source);
174             FileOutputStream fos = new FileOutputStream(target)) {
175            byte[] buffer = new byte[BUFFER_SIZE];
176            int bytesRead;
177            while ((bytesRead = fis.read(buffer)) != -1) {
178                fos.write(buffer, 0, bytesRead);
179            }
180        }
181    }
182
183    /**
184     * Split JSON file
185     */
186    private static List<File> splitJsonFile(File inputFile, File outputDir, int splitSizeMB) throws IOException {
187        List<File> splitFiles = new ArrayList<>();
188
189        // Read file content
190        String jsonContent = readFileContent(inputFile);
191        Map<?, ?> content = (Map<?, ?>) JSON.parseObject(jsonContent);
192        String createdBy = (String) content.get("createdBy");
193        String format = (String) content.get("format");
194
195        if (createdBy == null) {
196            logger.warn("JSON file does not contain 'createdBy' field, treating as regular file");
197            return splitFiles;
198        }
199
200        // Check for sqlflow-sharded format (new version)
201        if ("sqlflow-sharded".equals(format)) {
202            splitFiles.addAll(splitSqlflowShardedJson(inputFile, outputDir, splitSizeMB, content));
203        } else if (createdBy.toLowerCase().contains("sqldep") || createdBy.toLowerCase().contains("grabit")) {
204            splitFiles.addAll(splitSqldepGrabitJson(inputFile, outputDir, splitSizeMB, content, createdBy));
205        } else if (createdBy.toLowerCase().contains("sqlflow")) {
206            splitFiles.addAll(splitSqlflowJson(inputFile, outputDir, splitSizeMB, content));
207        } else {
208            logger.warn("Unknown JSON file type with createdBy: " + createdBy);
209        }
210
211        return splitFiles;
212    }
213
214    private static String readFileContent(File inputFile) {
215        return SQLUtil.getFileContent(inputFile);
216    }
217
218    /**
219     * Write JSON file
220     */
221    private static void writeJsonFile(File file, Object content) throws IOException {
222        String jsonString = JSON.toJSONString(content);
223        SQLUtil.writeToFile(file, jsonString);
224    }
225
226    /**
227     * Split sqldep or grabit format JSON file
228     */
229    private static List<File> splitSqldepGrabitJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> content, String createdBy) throws IOException {
230        List<File> splitFiles = new ArrayList<>();
231        List<Map<?, ?>> queries = (List<Map<?, ?>>) content.get("queries");
232        if (queries == null || queries.isEmpty()) {
233            logger.warn("No queries found in JSON file");
234            return splitFiles;
235        }
236
237        // Remove queries field, create metadata file
238        @SuppressWarnings("unchecked")
239        Map<String, Object> metadataContent = new LinkedHashMap<>((Map<String, Object>) content);
240        metadataContent.remove("queries");
241
242        String extension = getFileExtension(inputFile.getName());
243        File metadataFile = new File(outputDir, getFileNameWithoutExtension(inputFile.getName()) + "_" + 0 + "_" + 1 + (extension.isEmpty() ? "" : "." + extension));
244        writeJsonFile(metadataFile, metadataContent);
245        splitFiles.add(metadataFile);
246        logger.info("Created metadata file: " + metadataFile.getAbsolutePath());
247
248        // Split queries
249        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
250        int length = 0;
251        List<Map<?, ?>> temp = new ArrayList<>();
252        long startIndex = 1;
253        long endIndex = 1;
254
255        for (Map<?, ?> item : queries) {
256            temp.add(item);
257            String sourceCode = (String) item.get("sourceCode");
258            if (sourceCode != null) {
259                length += sourceCode.length();
260            }
261
262            if (length >= splitSizeBytes) {
263                File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
264                Map<String, Object> jsonObject = new LinkedHashMap<>();
265                jsonObject.put("createdBy", createdBy);
266                jsonObject.put("dbvendor", content.get("dbvendor"));
267                jsonObject.put("databases", new ArrayList<>());
268                jsonObject.put("queries", temp);
269                writeJsonFile(queryFile, jsonObject);
270                splitFiles.add(queryFile);
271                logger.info("Created query file: " + queryFile.getAbsolutePath());
272
273                temp.clear();
274                length = 0;
275                startIndex = endIndex;
276            }
277            endIndex++;
278        }
279
280        // Process remaining queries
281        if (!temp.isEmpty()) {
282            File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
283            Map<String, Object> jsonObject = new LinkedHashMap<>();
284            jsonObject.put("createdBy", createdBy);
285            jsonObject.put("dbvendor", content.get("dbvendor"));
286            jsonObject.put("databases", new ArrayList<>());
287            jsonObject.put("queries", temp);
288            writeJsonFile(queryFile, jsonObject);
289            splitFiles.add(queryFile);
290            logger.info("Created query file: " + queryFile.getAbsolutePath());
291        }
292
293        return splitFiles;
294    }
295
296    /**
297     * Split sqlflow format JSON file
298     */
299    private static List<File> splitSqlflowJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> content) throws IOException {
300        List<File> splitFiles = new ArrayList<>();
301        List<Map<?, ?>> queries = new ArrayList<>();
302
303        // Extract all queries
304        List<Map<?, ?>> servers = (List<Map<?, ?>>) content.get("servers");
305        if (servers != null) {
306            for (Map<?, ?> serverObject : servers) {
307                @SuppressWarnings("unchecked")
308                List<Map<?, ?>> serverQueries = (List<Map<?, ?>>) serverObject.get("queries");
309                if (serverQueries != null) {
310                    queries.addAll(serverQueries);
311                    serverObject.remove("queries");
312                }
313            }
314        }
315
316        if (queries.isEmpty()) {
317            logger.warn("No queries found in SQLFlow JSON file");
318            return splitFiles;
319        }
320
321        // Create metadata file
322        String extension = getFileExtension(inputFile.getName());
323        File metadataFile = new File(outputDir, getFileNameWithoutExtension(inputFile.getName()) + "_" + 0 + "_" + 1 + (extension.isEmpty() ? "" : "." + extension));
324        writeJsonFile(metadataFile, content);
325        splitFiles.add(metadataFile);
326        logger.info("Created metadata file: " + metadataFile.getAbsolutePath());
327
328        // Split queries
329        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
330        int length = 0;
331        List<Map<?, ?>> temp = new ArrayList<>();
332        long startIndex = 1;
333        long endIndex = 1;
334
335        for (Map<?, ?> item : queries) {
336            temp.add(item);
337            String sourceCode = (String) item.get("sourceCode");
338            if (sourceCode != null) {
339                length += sourceCode.length();
340            }
341
342            if (length >= splitSizeBytes) {
343                File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
344                Map<String, Object> jsonObject = new LinkedHashMap<>();
345                jsonObject.put("createdBy", "grabit v1.7.0");
346                jsonObject.put("databases", new ArrayList<>());
347                jsonObject.put("queries", temp);
348                writeJsonFile(queryFile, jsonObject);
349                splitFiles.add(queryFile);
350                logger.info("Created query file: " + queryFile.getAbsolutePath());
351
352                temp.clear();
353                length = 0;
354                startIndex = endIndex;
355            }
356            endIndex++;
357        }
358
359        // Process remaining queries
360        if (!temp.isEmpty()) {
361            File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
362            Map<String, Object> jsonObject = new LinkedHashMap<>();
363            jsonObject.put("createdBy", "grabit v1.7.0");
364            jsonObject.put("databases", new ArrayList<>());
365            jsonObject.put("queries", temp);
366            writeJsonFile(queryFile, jsonObject);
367            splitFiles.add(queryFile);
368            logger.info("Created query file: " + queryFile.getAbsolutePath());
369        }
370
371        return splitFiles;
372    }
373
374    //autogenerate
375    /**
376     * Split sqlflow-sharded format JSON file.
377     * Convert sqlflow-sharded to legacy sqlflow format:
378     * - catalog: if single db > splitSizeMB, keep as one file; else merge with next db until >= splitSizeMB
379     * - source: if single db < splitSizeMB, merge with next db; if > splitSizeMB, split by size keeping records complete
380     */
381    private static List<File> splitSqlflowShardedJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> manifest) throws IOException {
382        List<File> splitFiles = new ArrayList<>();
383        
384        File baseDir = inputFile.getParentFile();
385        if (baseDir == null) {
386            logger.warn("Cannot determine base directory from manifest file");
387            return splitFiles;
388        }
389        
390        String createdBy = (String) manifest.get("createdBy");
391        List<Map<?, ?>> servers = (List<Map<?, ?>>) manifest.get("servers");
392        if (servers == null || servers.isEmpty()) {
393            logger.warn("No servers found in sqlflow-sharded manifest");
394            return splitFiles;
395        }
396        
397        String extension = getFileExtension(inputFile.getName());
398        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
399        
400        // Collect all database info first
401        List<DatabaseCatalogInfo> allCatalogs = new ArrayList<>();
402        List<DatabaseSourceInfo> allSources = new ArrayList<>();
403        
404        for (Map<?, ?> server : servers) {
405            String serverName = (String) server.get("name");
406            String dbVendor = (String) server.get("dbVendor");
407            
408            List<Map<?, ?>> databases = (List<Map<?, ?>>) server.get("databases");
409            if (databases == null || databases.isEmpty()) {
410                continue;
411            }
412            
413            for (Map<?, ?> database : databases) {
414                String databaseName = (String) database.get("name");
415                String shardId = (String) database.get("shardId");
416                
417                // Collect catalog info
418                Map<?, ?> catalogInfo = (Map<?, ?>) database.get("catalog");
419                if (catalogInfo != null) {
420                    String catalogPath = (String) catalogInfo.get("path");
421                    File catalogFile = resolvePath(baseDir, catalogPath);
422                    if (catalogFile != null && catalogFile.exists()) {
423                        allCatalogs.add(new DatabaseCatalogInfo(serverName, dbVendor, databaseName, shardId, catalogFile));
424                    }
425                }
426                
427                // Collect source info
428                Map<?, ?> sourceInfo = (Map<?, ?>) database.get("source");
429                String sourceCompression = (String) manifest.get("sourceCompression");
430                if (sourceInfo != null) {
431                    String sourcePath = (String) sourceInfo.get("path");
432                    File sourceFile = resolvePath(baseDir, sourcePath);
433                    if (sourceFile != null && sourceFile.exists()) {
434                        allSources.add(new DatabaseSourceInfo(serverName, dbVendor, databaseName, shardId, sourceFile, sourceCompression));
435                    }
436                }
437            }
438        }
439        
440        // Process catalogs: merge if total < splitSizeMB, otherwise keep single
441        List<File> catalogFiles = processCatalogs(outputDir, inputFile, extension, allCatalogs, splitSizeBytes, createdBy);
442        splitFiles.addAll(catalogFiles);
443        
444        // Process sources: merge if total < splitSizeMB, split by size if > splitSizeMB
445        List<File> sourceFiles = processSources(outputDir, inputFile, extension, allSources, splitSizeBytes, createdBy);
446        splitFiles.addAll(sourceFiles);
447        
448        return splitFiles;
449    }
450    
451    /**
452     * Process catalogs: merge if total size < 2*splitSizeMB, otherwise keep single
453     */
454    private static List<File> processCatalogs(File outputDir, File inputFile, String extension,
455                                           List<DatabaseCatalogInfo> catalogs, long splitSizeBytes,
456                                           String createdBy) throws IOException {
457        List<File> result = new ArrayList<>();
458        
459        if (catalogs == null || catalogs.isEmpty()) {
460            return result;
461        }
462        
463        int catalogIndex = 1;
464        List<DatabaseCatalogInfo> batch = new ArrayList<>();
465        int currentSize = 0;
466        long maxBatchSize = splitSizeBytes * 2;
467        
468        for (DatabaseCatalogInfo catalog : catalogs) {
469            int catalogSize = (int) catalog.file.length();
470            
471            // If single catalog > splitSizeBytes (5MB), it must be alone
472            if (catalogSize > splitSizeBytes) {
473                // Save current batch first
474                if (!batch.isEmpty()) {
475                    File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
476                    if (batchFile != null) {
477                        result.add(batchFile);
478                        logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
479                    }
480                    catalogIndex++;
481                    batch = new ArrayList<>();
482                    currentSize = 0;
483                }
484                // Write this large catalog alone
485                File singleFile = createCatalogSingleFile(outputDir, inputFile, catalog, extension, catalogIndex, createdBy);
486                if (singleFile != null) {
487                    result.add(singleFile);
488                    logger.info("Created catalog file: " + singleFile.getAbsolutePath());
489                }
490                catalogIndex++;
491            } else {
492                // Check if adding this catalog would exceed 2*splitSize (10MB)
493                if (currentSize + catalogSize > maxBatchSize && !batch.isEmpty()) {
494                    // Save current batch
495                    File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
496                    if (batchFile != null) {
497                        result.add(batchFile);
498                        logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
499                    }
500                    catalogIndex++;
501                    batch = new ArrayList<>();
502                    currentSize = 0;
503                }
504                batch.add(catalog);
505                currentSize += catalogSize;
506            }
507        }
508        
509        // Save remaining batch
510        if (!batch.isEmpty()) {
511            File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
512            if (batchFile != null) {
513                result.add(batchFile);
514                logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
515            }
516        }
517        
518        return result;
519    }
520    
521    /**
522     * Process sources: accumulate all records, split by size > splitSizeMB
523     */
524    private static List<File> processSources(File outputDir, File inputFile, String extension,
525                                          List<DatabaseSourceInfo> sources, long splitSizeBytes,
526                                          String createdBy) throws IOException {
527        List<File> result = new ArrayList<>();
528        
529        if (sources == null || sources.isEmpty()) {
530            return result;
531        }
532        
533        int startIndex = 1;
534        List<Map<String, Object>> currentBatch = new ArrayList<>();
535        int currentSize = 0;
536        List<String> mergedFrom = new ArrayList<>();
537        
538        for (DatabaseSourceInfo source : sources) {
539            mergedFrom.add(source.shardId);
540            
541            List<Map<?, ?>> records = readSourceRecords(source.file, source.compression);
542            if (records == null || records.isEmpty()) {
543                continue;
544            }
545            
546            for (Map<?, ?> record : records) {
547                Map<String, Object> query = new LinkedHashMap<>();
548                query.put("database", source.databaseName);
549                query.put("schema", record.get("schema"));
550                query.put("name", record.get("name"));
551                query.put("type", record.get("type"));
552                query.put("sourceCode", record.get("sourceCode"));
553                if (record.containsKey("groupName")) {
554                    query.put("groupName", record.get("groupName"));
555                }
556                
557                Object sourceCode = record.get("sourceCode");
558                int recordSize = sourceCode != null ? sourceCode.toString().length() : 0;
559                
560                if (currentSize + recordSize > splitSizeBytes && !currentBatch.isEmpty()) {
561                    int endIndex = startIndex + currentBatch.size() - 1;
562                    File batchFile = createSourceQueryFileCommon(outputDir, inputFile, extension, 
563                            currentBatch, mergedFrom, startIndex, endIndex, createdBy);
564                    if (batchFile != null) {
565                        result.add(batchFile);
566                        logger.info("Created source batch file: " + batchFile.getAbsolutePath());
567                    }
568                    startIndex = endIndex + 1;
569                    currentBatch = new ArrayList<>();
570                    currentSize = 0;
571                }
572                
573                currentBatch.add(query);
574                currentSize += recordSize;
575            }
576        }
577        
578        // Save remaining batch
579        if (!currentBatch.isEmpty()) {
580            int endIndex = startIndex + currentBatch.size() - 1;
581            File batchFile = createSourceQueryFileCommon(outputDir, inputFile, extension, 
582                    currentBatch, mergedFrom, startIndex, endIndex, createdBy);
583            if (batchFile != null) {
584                result.add(batchFile);
585                logger.info("Created source batch file: " + batchFile.getAbsolutePath());
586            }
587        }
588        
589        return result;
590    }
591    
592    /**
593     * Calculate total size of source records
594     */
595    private static int calculateSourceSize(List<Map<?, ?>> records) {
596        if (records == null) return 0;
597        int size = 0;
598        for (Map<?, ?> record : records) {
599            Object sourceCode = record.get("sourceCode");
600            if (sourceCode != null) {
601                size += sourceCode.toString().length();
602            }
603        }
604        return size;
605    }
606    
607    /**
608     * Create catalog file for a batch of databases
609     */
610    private static File createCatalogBatchFile(File outputDir, File inputFile, String extension,
611                                            List<DatabaseCatalogInfo> batch, int batchIndex,
612                                            String createdBy) throws IOException {
613        if (batch == null || batch.isEmpty()) {
614            return null;
615        }
616        
617        Map<String, Object> sqlflow = new LinkedHashMap<>();
618        sqlflow.put("createdBy", createdBy != null ? createdBy : "sqlflow-ingester");
619        sqlflow.put("format", "sqlflow");
620        
621        // Add mergedFrom field to track original shardIds
622        List<String> mergedFrom = new ArrayList<>();
623        
624        List<Map<String, Object>> servers = new ArrayList<>();
625        
626        for (DatabaseCatalogInfo catalog : batch) {
627            mergedFrom.add(catalog.shardId);
628            
629            Map<String, Object> catalogContent = readCatalogContent(catalog.file);
630            if (catalogContent == null) continue;
631            
632            Map<String, Object> serverObj = new LinkedHashMap<>();
633            serverObj.put("name", catalog.serverName);
634            serverObj.put("dbVendor", catalog.dbVendor);
635            
636            Map<String, Object> dbObj = new LinkedHashMap<>();
637            dbObj.put("name", catalog.databaseName);
638            
639            copyCatalogFields(catalogContent, dbObj);
640            
641            List<Map<String, Object>> databases = new ArrayList<>();
642            databases.add(dbObj);
643            serverObj.put("databases", databases);
644            servers.add(serverObj);
645        }
646        
647        sqlflow.put("servers", servers);
648        sqlflow.put("databases", new ArrayList<>());
649        sqlflow.put("mergedFrom", mergedFrom);
650        
651        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_catalog_" + batchIndex +
652                (extension.isEmpty() ? "" : "." + extension);
653        
654        File outputFile = new File(outputDir, fileName);
655        writeJsonFile(outputFile, sqlflow);
656        return outputFile;
657    }
658    
659    /**
660     * Create catalog file for a single large database
661     */
662    private static File createCatalogSingleFile(File outputDir, File inputFile, DatabaseCatalogInfo catalog,
663                                             String extension, int catalogIndex, String createdBy) throws IOException {
664        Map<String, Object> catalogContent = readCatalogContent(catalog.file);
665        if (catalogContent == null) {
666            return null;
667        }
668        
669        Map<String, Object> sqlflow = new LinkedHashMap<>();
670        sqlflow.put("createdBy", createdBy != null ? createdBy : "sqlflow-ingester");
671        sqlflow.put("format", "sqlflow");
672        
673        Map<String, Object> serverObj = new LinkedHashMap<>();
674        serverObj.put("name", catalog.serverName);
675        serverObj.put("dbVendor", catalog.dbVendor);
676        
677        Map<String, Object> dbObj = new LinkedHashMap<>();
678        dbObj.put("name", catalog.databaseName);
679        copyCatalogFields(catalogContent, dbObj);
680        
681        List<Map<String, Object>> databases = new ArrayList<>();
682        databases.add(dbObj);
683        serverObj.put("databases", databases);
684        
685        List<Map<String, Object>> servers = new ArrayList<>();
686        servers.add(serverObj);
687        sqlflow.put("servers", servers);
688        sqlflow.put("databases", new ArrayList<>());
689        
690        // Add mergedFrom field to track original shardId
691        List<String> mergedFrom = new ArrayList<>();
692        mergedFrom.add(catalog.shardId);
693        sqlflow.put("mergedFrom", mergedFrom);
694        
695        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_catalog_" + catalogIndex +
696                (extension.isEmpty() ? "" : "." + extension);
697        
698        File outputFile = new File(outputDir, fileName);
699        writeJsonFile(outputFile, sqlflow);
700        return outputFile;
701    }
702    
703    /**
704     * Read catalog content from file
705     */
706    private static Map<String, Object> readCatalogContent(File catalogFile) throws IOException {
707        String content = readFileContent(catalogFile);
708        @SuppressWarnings("unchecked")
709        Map<String, Object> catalog = (Map<String, Object>) JSON.parseObject(content);
710        return catalog;
711    }
712    
713    /**
714     * Copy catalog fields to database object
715     */
716    private static void copyCatalogFields(Map<String, Object> catalog, Map<String, Object> dbObj) {
717        if (catalog.containsKey("schemas")) dbObj.put("schemas", catalog.get("schemas"));
718        if (catalog.containsKey("tables")) dbObj.put("tables", catalog.get("tables"));
719        if (catalog.containsKey("views")) dbObj.put("views", catalog.get("views"));
720        if (catalog.containsKey("procedures")) dbObj.put("procedures", catalog.get("procedures"));
721        if (catalog.containsKey("functions")) dbObj.put("functions", catalog.get("functions"));
722        if (catalog.containsKey("triggers")) dbObj.put("triggers", catalog.get("triggers"));
723        if (catalog.containsKey("packages")) dbObj.put("packages", catalog.get("packages"));
724        if (catalog.containsKey("synonyms")) dbObj.put("synonyms", catalog.get("synonyms"));
725    }
726    
727    /**
728     * Common method to create source query file
729     */
730    private static File createSourceQueryFileCommon(File outputDir, File inputFile, String extension,
731                                                  List<Map<String, Object>> queries, List<String> mergedFrom,
732                                                  int startIndex, int endIndex,
733                                                  String createdBy) throws IOException {
734        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_source_" + startIndex + "_" + endIndex +
735                (extension.isEmpty() ? "" : "." + extension);
736        
737        File queryFile = new File(outputDir, fileName);
738        
739        Map<String, Object> sqlflow = new LinkedHashMap<>();
740        sqlflow.put("createdBy", createdBy != null ? createdBy : "grabit v1.7.0");
741        sqlflow.put("databases", new ArrayList<>());
742        sqlflow.put("mergedFrom", mergedFrom);
743        
744        Map<String, Object> serverObj = new LinkedHashMap<>();
745        serverObj.put("name", "default");
746        serverObj.put("queries", queries);
747        
748        List<Map<String, Object>> servers = new ArrayList<>();
749        servers.add(serverObj);
750        sqlflow.put("servers", servers);
751        
752        writeJsonFile(queryFile, sqlflow);
753        return queryFile;
754    }
755    
756    /**
757     * Resolve path relative to base directory
758     */
759    private static File resolvePath(File baseDir, String path) {
760        if (path == null || path.isEmpty()) {
761            return null;
762        }
763        
764        File file = new File(path);
765        if (file.isAbsolute()) {
766            return file;
767        }
768        
769        return new File(baseDir, path);
770    }
771    
772    /**
773     * Read source records from JSONL file (supports gzip compression)
774     */
775    private static List<Map<?, ?>> readSourceRecords(File sourceFile, String compression) throws IOException {
776        if (sourceFile == null || !sourceFile.exists()) {
777            return null;
778        }
779        
780        List<Map<?, ?>> records = new ArrayList<>();
781        
782        if ("block".equals(compression)) {
783            // Read gzip compressed file
784            try (java.util.zip.GZIPInputStream gzis = new java.util.zip.GZIPInputStream(
785                    new java.io.FileInputStream(sourceFile))) {
786                java.io.BufferedReader reader = new java.io.BufferedReader(
787                        new java.io.InputStreamReader(gzis, "UTF-8"));
788                String line;
789                while ((line = reader.readLine()) != null) {
790                    if (!line.trim().isEmpty()) {
791                        Map<?, ?> record = (Map<?, ?>) JSON.parseObject(line);
792                        records.add(record);
793                    }
794                }
795            }
796        } else {
797            // Read plain JSONL file
798            String content = readFileContent(sourceFile);
799            String[] lines = content.split("\n");
800            for (String line : lines) {
801                if (!line.trim().isEmpty()) {
802                    Map<?, ?> record = (Map<?, ?>) JSON.parseObject(line);
803                    records.add(record);
804                }
805            }
806        }
807        
808        return records;
809    }
810    
811    /**
812     * Helper class for catalog info
813     */
814    private static class DatabaseCatalogInfo {
815        String serverName;
816        String dbVendor;
817        String databaseName;
818        String shardId;
819        File file;
820        
821        DatabaseCatalogInfo(String serverName, String dbVendor, String databaseName, String shardId, File file) {
822            this.serverName = serverName;
823            this.dbVendor = dbVendor;
824            this.databaseName = databaseName;
825            this.shardId = shardId;
826            this.file = file;
827        }
828    }
829    
830    /**
831     * Helper class for source info
832     */
833    private static class DatabaseSourceInfo {
834        String serverName;
835        String dbVendor;
836        String databaseName;
837        String shardId;
838        File file;
839        String compression;
840        
841        DatabaseSourceInfo(String serverName, String dbVendor, String databaseName, String shardId, File file, String compression) {
842            this.serverName = serverName;
843            this.dbVendor = dbVendor;
844            this.databaseName = databaseName;
845            this.shardId = shardId;
846            this.file = file;
847            this.compression = compression;
848        }
849    }
850
851    /**
852     * Create query file
853     */
854    private static File createQueryFile(File outputDir, File inputFile, long startIndex, long endIndex, String extension) {
855        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_" + startIndex + "_" + endIndex + (extension.isEmpty() ? "" : "." + extension);
856        return new File(outputDir, fileName);
857    }
858
859    private static Long getLastLineFromCurrentPos(RandomAccessFile raf, long currentPos, long fileLength, long splitSizeBytes, EDbVendor vendor) throws IOException {
860        raf.seek(currentPos);
861
862        long remainingBytes = fileLength - currentPos;
863        long readSize = Math.min(splitSizeBytes, remainingBytes);
864
865        byte[] buffer = new byte[(int) readSize];
866        int bytesRead = raf.read(buffer);
867        if (bytesRead <= 0) {
868            return null;
869        }
870
871        String partialContent = new String(buffer, 0, bytesRead, "UTF-8");
872        if (partialContent.isEmpty()) {
873            return null;
874        }
875
876        TGSqlParser parser = new TGSqlParser(vendor);
877        parser.sqltext = partialContent;
878        parser.getrawsqlstatements();
879
880        long lastValidatedLineNo = parser.getLastLineNoOfLastStatementBeenValidated();
881        return lastValidatedLineNo;
882    }
883
884    /**
885     * Find end position of Nth line from specified byte position
886     * Supports multiple newline formats: \n (Unix/Linux), \r\n (Windows), \r (Old Mac)
887     * Uses buffer for batch reading to improve performance
888     */
889    private static long findLineEndPosition(RandomAccessFile raf, long startPos, long lineCount) throws IOException {
890        raf.seek(startPos);
891        long newlineCount = 0;
892        long fileLength = raf.length();
893
894        byte[] buffer = new byte[BUFFER_SIZE];
895        long currentPos = startPos;
896
897        while (currentPos < fileLength) {
898            int toRead = (int) Math.min(buffer.length, fileLength - currentPos);
899            int bytesRead = raf.read(buffer, 0, toRead);
900            if (bytesRead <= 0) {
901                break;
902            }
903
904            for (int i = 0; i < bytesRead; i++) {
905                byte b = buffer[i];
906
907                if (b == '\n') {
908                    newlineCount++;
909                    if (newlineCount == lineCount) {
910                        return currentPos + i + 1;
911                    }
912                } else if (b == '\r') {
913                    if (i + 1 < bytesRead) {
914                        if (buffer[i + 1] == '\n') {
915                            newlineCount++;
916                            if (newlineCount == lineCount) {
917                                return currentPos + i + 2;
918                            }
919                            i++;
920                        } else {
921                            newlineCount++;
922                            if (newlineCount == lineCount) {
923                                return currentPos + i + 1;
924                            }
925                        }
926                    } else {
927                        long savedPos = raf.getFilePointer();
928                        int nextByte = raf.read();
929
930                        if (nextByte == '\n') {
931                            newlineCount++;
932                            if (newlineCount == lineCount) {
933                                return currentPos + i + 2;
934                            }
935                            currentPos++;
936                        } else {
937                            newlineCount++;
938                            if (newlineCount == lineCount) {
939                                return currentPos + i + 1;
940                            }
941                            if (nextByte != -1) {
942                                raf.seek(savedPos);
943                            }
944                        }
945                    }
946                }
947            }
948
949            currentPos += bytesRead;
950        }
951
952        return fileLength;
953    }
954
955    /**
956     * Split file by byte position
957     */
958    private static File splitFileByPosition(File outputDir, File inputFile, long startPos, long endPos, int fileIndex, long startLineNo, long endLineNo) throws IOException {
959        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_" + fileIndex + "_" + startLineNo + "_" + endLineNo + "." + getFileExtension(inputFile.getName());
960        File outputFile = new File(outputDir, fileName);
961
962        try (RandomAccessFile rafRead = new RandomAccessFile(inputFile, "r");
963             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(outputFile), java.nio.charset.StandardCharsets.UTF_8))) {
964
965            rafRead.seek(startPos);
966            long bytesToRead = endPos - startPos;
967            long bytesRemaining = bytesToRead;
968
969            byte[] buffer = new byte[BUFFER_SIZE];
970
971            while (bytesRemaining > 0) {
972                int toRead = (int) Math.min(buffer.length, bytesRemaining);
973                int bytesRead = rafRead.read(buffer, 0, toRead);
974                if (bytesRead <= 0) {
975                    break;
976                }
977
978                String chunk = new String(buffer, 0, bytesRead, java.nio.charset.StandardCharsets.UTF_8);
979                writer.write(chunk);
980                bytesRemaining -= bytesRead;
981            }
982
983            logger.info("split file " + inputFile.getName() + " (index: " + fileIndex + ", lines: " + startLineNo + "-" + endLineNo + ") from byte " + startPos + " to " + endPos + " to " + outputFile.getName());
984
985            return outputFile;
986        }
987    }
988}