package gudusoft.gsqlparser.util;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.util.json.JSON;

import java.io.*;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * Splits large files into smaller chunks for parallel processing.
 *
 * <p>Supports two categories of input:
 * <ul>
 *   <li><b>Plain SQL files</b> — split by statement boundaries using the SQL
 *       parser to find safe cut points.</li>
 *   <li><b>JSON metadata files</b> — split by format (sqldep/grabit, sqlflow,
 *       sqlflow-sharded), converting each into smaller self-contained JSON files
 *       that the re-reader can consume independently.</li>
 * </ul>
 *
 * <h3>JSON format routing</h3>
 * splitJsonFile() routes to the appropriate splitter based on
 * {@code createdBy} and {@code format}:
 * <ul>
 *   <li>{@code format: "sqlflow-sharded"} → splitSqlflowShardedJson()</li>
 *   <li>{@code createdBy} contains "sqldep" or "grabit" → splitSqldepGrabitJson()</li>
 *   <li>{@code createdBy} contains "sqlflow" → splitSqlflowJson()</li>
 * </ul>
 */
public class FileSplitter {

    private static final int MAX_CONSECUTIVE_FAILURES = 3;
    private static final int BUFFER_SIZE = 8192;
    private static final Logger logger = LoggerFactory.getLogger(FileSplitter.class);

    /**
     * Split large file
     *
     * @param inputFile   Input file
     * @param outputDir   Output directory
     * @param splitSizeMB Split size (MB)
     * @param dbVendor    Database vendor
     * @return List of split files
     * @throws IOException IO exception
     */
    public static List<File> splitFile(File inputFile, File outputDir, int splitSizeMB, EDbVendor dbVendor) throws IOException {
        List<File> splitFiles = new ArrayList<>();
        if (!inputFile.exists()) {
            throw new FileNotFoundException("Input file does not exist: " + inputFile.getAbsolutePath());
        }

        if (!outputDir.exists() && !outputDir.mkdirs()) {
            throw new IOException("Failed to create output directory: " + outputDir.getAbsolutePath());
        }

        // Check if it's a JSON file
        if (isJsonFile(inputFile)) {
            return splitJsonFile(inputFile, outputDir, splitSizeMB);
        }

        // Process regular SQL file
        long splitSizeBytes = splitSizeMB * 1024 * 1024L;

        try (RandomAccessFile raf = new RandomAccessFile(inputFile, "r")) {
            long fileLength = raf.length();
            long currentPos = 0;
            long chunkStartPos = 0;
            int consecutiveFailures = 0;
            int fileIndex = 1;
            long globalStartLineNo = 1;

            while (currentPos < fileLength) {
                Long lastValidatedLineNo = getLastLineFromCurrentPos(raf, currentPos, fileLength, splitSizeBytes, dbVendor);
                if (lastValidatedLineNo == null) break;

                while (lastValidatedLineNo != null && lastValidatedLineNo == -1) {
                    consecutiveFailures++;
                    logger.warn("No validated statement found at position " + currentPos + "  (consecutive failures: " + consecutiveFailures + ")");

                    long incrementSize = Math.min(splitSizeBytes, fileLength - currentPos);
                    long previousPos = currentPos;
                    currentPos = Math.min(currentPos + incrementSize, fileLength);

                    if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
                        logger.error("Reached maximum consecutive failures (" + MAX_CONSECUTIVE_FAILURES + "), skipping this segment and moving to next");
                        long skipSize = splitSizeBytes * (consecutiveFailures + 1);
                        currentPos = Math.min(previousPos + skipSize, fileLength);
                        chunkStartPos = currentPos;
                        consecutiveFailures = 0; // Reset consecutive failure counter
                        logger.info("Skipped to position " + currentPos + ", continuing with next segment");
                        break;
                    }

                    lastValidatedLineNo = getLastLineFromCurrentPos(raf, currentPos, fileLength, splitSizeBytes * (consecutiveFailures + 1), dbVendor);
                }

                if (lastValidatedLineNo == null) {
                    break;
                }

                if (lastValidatedLineNo == -1) {
                    continue;
                }

                consecutiveFailures = 0;
                long endPos = findLineEndPosition(raf, currentPos, lastValidatedLineNo);
                long globalEndLineNo = globalStartLineNo + lastValidatedLineNo - 1;

                File splitFile = splitFileByPosition(outputDir, inputFile, chunkStartPos, endPos, fileIndex, globalStartLineNo, globalEndLineNo);
                splitFiles.add(splitFile);

                chunkStartPos = endPos;
                currentPos = endPos;
                globalStartLineNo = globalEndLineNo + 1;
                fileIndex++;

                if (currentPos >= fileLength) {
                    break;
                }
            }
        }

        return splitFiles;
    }

    /**
     * Check if file is a JSON file
     */
    private static boolean isJsonFile(File file) {
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            int c;
            // Skip whitespace characters
            while ((c = reader.read()) != -1) {
                if (!Character.isWhitespace(c)) {
                    // Check if first non-whitespace character is '{'
                    return c == '{';
                }
            }
        } catch (IOException e) {
            logger.error("Error reading file to check if it's JSON: " + file.getAbsolutePath(), e);
        }
        return false;
    }

    /**
     * Get file extension
     */
    private static String getFileExtension(String fileName) {
        int lastDotIndex = fileName.lastIndexOf('.');
        if (lastDotIndex == -1) {
            return "";
        }
        return fileName.substring(lastDotIndex + 1);
    }

    /**
     * Get file name without extension
     */
    private static String getFileNameWithoutExtension(String fileName) {
        int lastDotIndex = fileName.lastIndexOf('.');
        if (lastDotIndex == -1) {
            return fileName;
        }
        return fileName.substring(0, lastDotIndex);
    }

    /**
     * Move file
     */
    private static void moveFile(File source, File target) throws IOException {
        if (!source.exists()) {
            throw new FileNotFoundException("Source file does not exist: " + source.getAbsolutePath());
        }
        if (target.exists() && !target.delete()) {
            throw new IOException("Failed to delete existing target file: " + target.getAbsolutePath());
        }
        if (!source.renameTo(target)) {
            // If renameTo fails, try copy and delete
            copyFile(source, target);
            if (!source.delete()) {
                throw new IOException("Failed to delete source file after copy: " + source.getAbsolutePath());
            }
        }
    }

    /**
     * Copy file
     */
    private static void copyFile(File source, File target) throws IOException {
        try (FileInputStream fis = new FileInputStream(source);
             FileOutputStream fos = new FileOutputStream(target)) {
            byte[] buffer = new byte[BUFFER_SIZE];
            int bytesRead;
            while ((bytesRead = fis.read(buffer)) != -1) {
                fos.write(buffer, 0, bytesRead);
            }
        }
    }

    /**
         * Split a JSON metadata file by routing to the format-specific splitter.
         *
         * <p>Routing logic:
         * <ul>
         *   <li>{@code format: "sqlflow-sharded"} → splitSqlflowShardedJson()
         *       (after version-gating via isSupportedSqlflowSharded)</li>
         *   <li>{@code createdBy} contains "sqldep" or "grabit" → splitSqldepGrabitJson()</li>
         *   <li>{@code createdBy} contains "sqlflow" (and no format key) → splitSqlflowJson()</li>
         * </ul>
         *
         * @param inputFile   the JSON file to split
         * @param outputDir   the directory to write split files
         * @param splitSizeMB target split size in megabytes
         * @return list of generated split files
         */
        private static List<File> splitJsonFile(File inputFile, File outputDir, int splitSizeMB) throws IOException {
        List<File> splitFiles = new ArrayList<>();

        // Read file content
        String jsonContent = readFileContent(inputFile);
        Map<?, ?> content = (Map<?, ?>) JSON.parseObject(jsonContent);
        String createdBy = (String) content.get("createdBy");
        String format = (String) content.get("format");

        if (createdBy == null) {
            logger.warn("JSON file does not contain 'createdBy' field, treating as regular file");
            return splitFiles;
        }

        // Check for sqlflow-sharded format (new version)
        if ("sqlflow-sharded".equals(format)) {
            if (!gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader
                    .isSupportedSqlflowSharded(jsonContent)) {
                // Unknown future formatVersion: do not split/rewrite a manifest we
                // may mis-read into legacy files that strip the version.
                logger.warn("Unsupported sqlflow-sharded formatVersion in " + inputFile
                        + "; not split.");
                return splitFiles;
            }
            splitFiles.addAll(splitSqlflowShardedJson(inputFile, outputDir, splitSizeMB, content));
        } else if (createdBy.toLowerCase().contains("sqldep") || createdBy.toLowerCase().contains("grabit")) {
            splitFiles.addAll(splitSqldepGrabitJson(inputFile, outputDir, splitSizeMB, content, createdBy));
        } else if (createdBy.toLowerCase().contains("sqlflow")) {
            splitFiles.addAll(splitSqlflowJson(inputFile, outputDir, splitSizeMB, content));
        } else {
            logger.warn("Unknown JSON file type with createdBy: " + createdBy);
        }

        return splitFiles;
    }

    private static String readFileContent(File inputFile) {
        return SQLUtil.getFileContent(inputFile);
    }

    /**
     * Write JSON file
     */
    private static void writeJsonFile(File file, Object content) throws IOException {
        try (FileWriter writer = new FileWriter(file)) {
            JSON.toJSONString(content, writer);
        }
    }

    /**
         * Split a sqldep or grabit format JSON file.
         *
         * <p>These formats have {@code queries[]} at the root level. The splitter:
         * <ul>
         *   <li>Creates a metadata file (everything except queries) as the first
         *       output file.</li>
         *   <li>Splits the queries into batches by cumulative sourceCode size,
         *       preserving the original {@code createdBy} so the re-reader
         *       recognizes the format via {@code MetadataReader.isGrabit()}.</li>
         * </ul>
         *
         * @param inputFile   the original JSON file
         * @param outputDir   the directory to write split files
         * @param splitSizeMB target split size in megabytes
         * @param content     the parsed JSON content
         * @param createdBy   the original createdBy string
         * @return list of generated split files
         */
        private static List<File> splitSqldepGrabitJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> content, String createdBy) throws IOException {
        List<File> splitFiles = new ArrayList<>();
        List<Map<?, ?>> queries = (List<Map<?, ?>>) content.get("queries");
        if (queries == null || queries.isEmpty()) {
            logger.warn("No queries found in JSON file");
            return splitFiles;
        }

        // Remove queries field, create metadata file
        @SuppressWarnings("unchecked")
        Map<String, Object> metadataContent = new LinkedHashMap<>((Map<String, Object>) content);
        metadataContent.remove("queries");

        String extension = getFileExtension(inputFile.getName());
        File metadataFile = new File(outputDir, getFileNameWithoutExtension(inputFile.getName()) + "_" + 0 + "_" + 1 + (extension.isEmpty() ? "" : "." + extension));
        writeJsonFile(metadataFile, metadataContent);
        splitFiles.add(metadataFile);
        logger.info("Created metadata file: " + metadataFile.getAbsolutePath());

        // Split queries
        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
        int length = 0;
        List<Map<?, ?>> temp = new ArrayList<>();
        long startIndex = 1;
        long endIndex = 1;

        for (Map<?, ?> item : queries) {
            temp.add(item);
            String sourceCode = (String) item.get("sourceCode");
            if (sourceCode != null) {
                length += sourceCode.length();
            }

            if (length >= splitSizeBytes) {
                File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
                Map<String, Object> jsonObject = new LinkedHashMap<>();
                jsonObject.put("createdBy", createdBy);
                jsonObject.put("dbvendor", content.get("dbvendor"));
                jsonObject.put("databases", new ArrayList<>());
                jsonObject.put("queries", temp);
                writeJsonFile(queryFile, jsonObject);
                splitFiles.add(queryFile);
                logger.info("Created query file: " + queryFile.getAbsolutePath());

                temp.clear();
                length = 0;
                startIndex = endIndex;
            }
            endIndex++;
        }

        // Process remaining queries
        if (!temp.isEmpty()) {
            File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
            Map<String, Object> jsonObject = new LinkedHashMap<>();
            jsonObject.put("createdBy", createdBy);
            jsonObject.put("dbvendor", content.get("dbvendor"));
            jsonObject.put("databases", new ArrayList<>());
            jsonObject.put("queries", temp);
            writeJsonFile(queryFile, jsonObject);
            splitFiles.add(queryFile);
            logger.info("Created query file: " + queryFile.getAbsolutePath());
        }

        return splitFiles;
    }

    /**
         * Split an old sqlflow format JSON file.
         *
         * <p>The old sqlflow format has {@code servers[].queries[]} structure with
         * no {@code format} key. The splitter:
         * <ul>
         *   <li>Extracts all queries from all servers, removing them from the
         *       metadata.</li>
         *   <li>Creates a metadata file with the servers (minus queries) as the
         *       first output file.</li>
         *   <li>Splits the queries into batches by cumulative sourceCode size,
         *       writing each batch as a grabit-format file ({@code createdBy: "grabit
         *       v1.7.0"}, {@code queries[]} at root level) so the re-reader
         *       recognizes it via {@code MetadataReader.isGrabit()}.</li>
         * </ul>
         *
         * @param inputFile   the original JSON file
         * @param outputDir   the directory to write split files
         * @param splitSizeMB target split size in megabytes
         * @param content     the parsed JSON content
         * @return list of generated split files
         */
        private static List<File> splitSqlflowJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> content) throws IOException {
        List<File> splitFiles = new ArrayList<>();
        List<Map<?, ?>> queries = new ArrayList<>();

        // Extract all queries
        List<Map<?, ?>> servers = (List<Map<?, ?>>) content.get("servers");
        if (servers != null) {
            for (Map<?, ?> serverObject : servers) {
                @SuppressWarnings("unchecked")
                List<Map<?, ?>> serverQueries = (List<Map<?, ?>>) serverObject.get("queries");
                if (serverQueries != null) {
                    queries.addAll(serverQueries);
                    serverObject.remove("queries");
                }
            }
        }

        if (queries.isEmpty()) {
            logger.warn("No queries found in SQLFlow JSON file");
            return splitFiles;
        }

        // Create metadata file
        String extension = getFileExtension(inputFile.getName());
        File metadataFile = new File(outputDir, getFileNameWithoutExtension(inputFile.getName()) + "_" + 0 + "_" + 1 + (extension.isEmpty() ? "" : "." + extension));
        writeJsonFile(metadataFile, content);
        splitFiles.add(metadataFile);
        logger.info("Created metadata file: " + metadataFile.getAbsolutePath());

        // Split queries
        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
        int length = 0;
        List<Map<?, ?>> temp = new ArrayList<>();
        long startIndex = 1;
        long endIndex = 1;

        for (Map<?, ?> item : queries) {
            temp.add(item);
            String sourceCode = (String) item.get("sourceCode");
            if (sourceCode != null) {
                length += sourceCode.length();
            }

            if (length >= splitSizeBytes) {
                File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
                Map<String, Object> jsonObject = new LinkedHashMap<>();
                jsonObject.put("createdBy", "grabit v1.7.0");
                jsonObject.put("queries", temp);
                writeJsonFile(queryFile, jsonObject);
                splitFiles.add(queryFile);
                logger.info("Created query file: " + queryFile.getAbsolutePath());

                temp.clear();
                length = 0;
                startIndex = endIndex;
            }
            endIndex++;
        }

        // Process remaining queries
        if (!temp.isEmpty()) {
            File queryFile = createQueryFile(outputDir, inputFile, startIndex, endIndex, extension);
            Map<String, Object> jsonObject = new LinkedHashMap<>();
            jsonObject.put("createdBy", "grabit v1.7.0");
            jsonObject.put("queries", temp);
            writeJsonFile(queryFile, jsonObject);
            splitFiles.add(queryFile);
            logger.info("Created query file: " + queryFile.getAbsolutePath());
        }

        return splitFiles;
    }

    //autogenerate
    /**
         * Split a sqlflow-sharded manifest into multiple legacy sqlflow files suitable
         * for parallel processing.
         *
         * <h3>Format conversion</h3>
         * Input:  sqlflow-sharded ({@code format: "sqlflow-sharded"}, separate
         *         catalog/*.catalog.json and source/*.source.jsonl files).
         * Output: old sqlflow format (no {@code format} key, {@code createdBy}
         *         containing "sqlflow", catalog and source inlined into
         *         {@code servers[].databases[]} / {@code servers[].queries[]}).
         *
         * <p>The conversion is necessary because the re-reader identifies the old
         * sqlflow format via {@code MetadataReader.isSqlflow()}, which requires:
         * <ul>
         *   <li>No {@code format} key present</li>
         *   <li>{@code createdBy} contains "sqlflow"</li>
         * </ul>
         *
         * <h3>Split strategy</h3>
         * <ul>
         *   <li>Catalog: if a single database's catalog exceeds splitSizeMB, keep it
         *       as one file; otherwise merge consecutive databases until the batch
         *       reaches splitSizeMB.</li>
         *   <li>Source: if a single database's source is below splitSizeMB, merge with
         *       the next; if above, split by record count keeping individual records
         *       complete.</li>
         * </ul>
         *
         * <h3>Limitations</h3>
         * Only catalog-topology (servers[].databases[]) is supported. Schema-topology
         * (servers[].schemas[]) exports are skipped — the caller falls back to
         * whole-file processing.
         *
         * @param inputFile   the original manifest file
         * @param outputDir   the directory to write split files
         * @param splitSizeMB target split size in megabytes
         * @param manifest    the parsed manifest JSON
         * @return list of generated split files
         */
        private static List<File> splitSqlflowShardedJson(File inputFile, File outputDir, int splitSizeMB, Map<?, ?> manifest) throws IOException {
        List<File> splitFiles = new ArrayList<>();
        
        File baseDir = inputFile.getParentFile();
        if (baseDir == null) {
            logger.warn("Cannot determine base directory from manifest file");
            return splitFiles;
        }
        
        String createdBy = (String) manifest.get("createdBy");
        List<Map<?, ?>> servers = (List<Map<?, ?>>) manifest.get("servers");
        if (servers == null || servers.isEmpty()) {
            logger.warn("No servers found in sqlflow-sharded manifest");
            return splitFiles;
        }
        
        String extension = getFileExtension(inputFile.getName());
        long splitSizeBytes = splitSizeMB * 1024 * 1024L;
        
        // Collect all database info first
        List<DatabaseCatalogInfo> allCatalogs = new ArrayList<>();
        List<DatabaseSourceInfo> allSources = new ArrayList<>();
        
        for (Map<?, ?> server : servers) {
            String serverName = (String) server.get("name");
            String dbVendor = (String) server.get("dbVendor");
            
            // NOTE: schema-topology exports (servers[].schemas[]) are
            // intentionally NOT split here. This splitter's output writers emit
            // catalog-topology (databases[]-wrapped) shards; emitting flat
            // schema shards through them yields a manifest legacy readers load
            // as 0 tables. Rather than produce a broken split, we skip schema
            // shards so the caller falls back to whole-file processing, which
            // the readers now handle correctly (SqlflowSharded* schema-topology
            // support). Splitting schema-topology manifests needs schema-aware
            // output writers + a round-trip test — tracked as a follow-up.
            List<Map<?, ?>> databases = (List<Map<?, ?>>) server.get("databases");
            if (databases == null || databases.isEmpty()) {
                continue;
            }
            
            for (Map<?, ?> database : databases) {
                String databaseName = (String) database.get("name");
                String shardId = (String) database.get("shardId");
                
                // Collect catalog info
                Map<?, ?> catalogInfo = (Map<?, ?>) database.get("catalog");
                if (catalogInfo != null) {
                    String catalogPath = (String) catalogInfo.get("path");
                    File catalogFile = resolvePath(baseDir, catalogPath);
                    if (catalogFile != null && catalogFile.exists()) {
                        allCatalogs.add(new DatabaseCatalogInfo(serverName, dbVendor, databaseName, shardId, catalogFile));
                    }
                }
                
                // Collect source info
                Map<?, ?> sourceInfo = (Map<?, ?>) database.get("source");
                String sourceCompression = (String) manifest.get("sourceCompression");
                if (sourceInfo != null) {
                    String sourcePath = (String) sourceInfo.get("path");
                    File sourceFile = resolvePath(baseDir, sourcePath);
                    if (sourceFile != null && sourceFile.exists()) {
                        allSources.add(new DatabaseSourceInfo(serverName, dbVendor, databaseName, shardId, sourceFile, sourceCompression));
                    }
                }
            }
        }
        
        // Process catalogs: merge if total < splitSizeMB, otherwise keep single
        List<File> catalogFiles = processCatalogs(outputDir, inputFile, extension, allCatalogs, splitSizeBytes, createdBy);
        splitFiles.addAll(catalogFiles);
        
        // Process sources: merge if total < splitSizeMB, split by size if > splitSizeMB
        List<File> sourceFiles = processSources(outputDir, inputFile, extension, allSources, splitSizeBytes, createdBy);
        splitFiles.addAll(sourceFiles);
        
        return splitFiles;
    }
    
    /**
         * Process catalog shards: batch small catalogs together, keep large ones
         * alone.
         *
         * <p>Strategy:
         * <ul>
         *   <li>If a single catalog exceeds splitSizeBytes, write it as a standalone
         *       file via {@link #createCatalogSingleFile}.</li>
         *   <li>Otherwise, accumulate catalogs into a batch. Flush the batch when
         *       adding the next catalog would exceed maxBatchSize (2× splitSizeBytes).</li>
         *   <li>Any remaining batch is flushed at the end.</li>
         * </ul>
         *
         * <p>All output files use the old sqlflow format (no {@code format} key,
         * {@code servers[].databases[]}) so the re-reader recognizes them via
         * {@code MetadataReader.isSqlflow()}.
         *
         * @param outputDir     the directory to write output files
         * @param inputFile     the original manifest file (used for naming)
         * @param extension     the file extension
         * @param catalogs      the list of catalog shards to process
         * @param splitSizeBytes target split size in bytes
         * @param createdBy     the createdBy string (must contain "sqlflow")
         * @return list of generated catalog files
         */
        private static List<File> processCatalogs(File outputDir, File inputFile, String extension,
                                           List<DatabaseCatalogInfo> catalogs, long splitSizeBytes,
                                           String createdBy) throws IOException {
        List<File> result = new ArrayList<>();
        
        if (catalogs == null || catalogs.isEmpty()) {
            return result;
        }
        
        int catalogIndex = 1;
        List<DatabaseCatalogInfo> batch = new ArrayList<>();
        int currentSize = 0;
        long maxBatchSize = splitSizeBytes * 2;
        
        for (DatabaseCatalogInfo catalog : catalogs) {
            int catalogSize = (int) catalog.file.length();
            
            // If single catalog > splitSizeBytes (5MB), it must be alone
            if (catalogSize > splitSizeBytes) {
                // Save current batch first
                if (!batch.isEmpty()) {
                    File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
                    if (batchFile != null) {
                        result.add(batchFile);
                        logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
                    }
                    catalogIndex++;
                    batch = new ArrayList<>();
                    currentSize = 0;
                }
                // Write this large catalog alone
                File singleFile = createCatalogSingleFile(outputDir, inputFile, catalog, extension, catalogIndex, createdBy);
                if (singleFile != null) {
                    result.add(singleFile);
                    logger.info("Created catalog file: " + singleFile.getAbsolutePath());
                }
                catalogIndex++;
            } else {
                // Check if adding this catalog would exceed 2*splitSize (10MB)
                if (currentSize + catalogSize > maxBatchSize && !batch.isEmpty()) {
                    // Save current batch
                    File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
                    if (batchFile != null) {
                        result.add(batchFile);
                        logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
                    }
                    catalogIndex++;
                    batch = new ArrayList<>();
                    currentSize = 0;
                }
                batch.add(catalog);
                currentSize += catalogSize;
            }
        }
        
        // Save remaining batch
        if (!batch.isEmpty()) {
            File batchFile = createCatalogBatchFile(outputDir, inputFile, extension, batch, catalogIndex, createdBy);
            if (batchFile != null) {
                result.add(batchFile);
                logger.info("Created catalog batch file: " + batchFile.getAbsolutePath());
            }
        }
        
        return result;
    }
    
    /**
         * Process source shards: read all records from JSONL source files, split them
         * into batches by cumulative sourceCode size.
         *
         * <p>Each source record is enriched with its database name and preserved
         * fields ({@code schema}, {@code name}, {@code type}, {@code sourceCode},
         * {@code groupName}, {@code sourceUnavailable}, {@code sourceUnavailableReason})
         * so the downstream analyzer has the full per-query context.
         *
         * <p>Output files use the old sqlflow format ({@code servers[].queries[]},
         * no {@code format} key) with {@code createdBy} set to the caller-provided
         * value. The re-reader must contain "sqlflow" to be recognized by
         * {@code MetadataReader.isSqlflow()}.
         *
         * @param outputDir      the directory to write output files
         * @param inputFile      the original manifest file (used for naming)
         * @param extension      the file extension
         * @param sources        the list of source shards to process
         * @param splitSizeBytes target split size in bytes
         * @param createdBy      the createdBy string (must contain "sqlflow")
         * @return list of generated source files
         */
        private static List<File> processSources(File outputDir, File inputFile, String extension,
                                          List<DatabaseSourceInfo> sources, long splitSizeBytes,
                                          String createdBy) throws IOException {
        List<File> result = new ArrayList<>();
        
        if (sources == null || sources.isEmpty()) {
            return result;
        }
        
        int startIndex = 1;
        List<SourceQueryEntry> currentBatch = new ArrayList<>();
        int currentSize = 0;
        List<String> mergedFrom = new ArrayList<>();

        for (DatabaseSourceInfo source : sources) {
            mergedFrom.add(source.shardId);

            List<Map<?, ?>> records = readSourceRecords(source.file, source.compression);
            if (records == null || records.isEmpty()) {
                continue;
            }

            for (Map<?, ?> record : records) {
                Map<String, Object> query = new LinkedHashMap<>();
                query.put("database", source.databaseName);
                query.put("schema", record.get("schema"));
                query.put("name", record.get("name"));
                query.put("type", record.get("type"));
                query.put("sourceCode", record.get("sourceCode"));
                if (record.containsKey("groupName")) {
                    query.put("groupName", record.get("groupName"));
                }
                // Preserve the availability flag through the split, or the
                // downstream analyzer cannot tell an unavailable (encrypted /
                // permission-denied) record from a real one and would parse its
                // placeholder text as live SQL.
                if (record.containsKey("sourceUnavailable")) {
                    query.put("sourceUnavailable", record.get("sourceUnavailable"));
                }
                if (record.containsKey("sourceUnavailableReason")) {
                    query.put("sourceUnavailableReason", record.get("sourceUnavailableReason"));
                }

                Object sourceCode = record.get("sourceCode");
                int recordSize = sourceCode != null ? sourceCode.toString().length() : 0;

                if (currentSize + recordSize > splitSizeBytes && !currentBatch.isEmpty()) {
                    int endIndex = startIndex + currentBatch.size() - 1;
                    File batchFile = createSourceQueryFileCommon(outputDir, inputFile, extension,
                            currentBatch, mergedFrom, startIndex, endIndex, createdBy);
                    if (batchFile != null) {
                        result.add(batchFile);
                        logger.info("Created source batch file: " + batchFile.getAbsolutePath());
                    }
                    startIndex = endIndex + 1;
                    currentBatch = new ArrayList<>();
                    currentSize = 0;
                }

                currentBatch.add(new SourceQueryEntry(source.serverName, source.dbVendor, query));
                currentSize += recordSize;
            }
        }

        // Save remaining batch
        if (!currentBatch.isEmpty()) {
            int endIndex = startIndex + currentBatch.size() - 1;
            File batchFile = createSourceQueryFileCommon(outputDir, inputFile, extension,
                    currentBatch, mergedFrom, startIndex, endIndex, createdBy);
            if (batchFile != null) {
                result.add(batchFile);
                logger.info("Created source batch file: " + batchFile.getAbsolutePath());
            }
        }

        return result;
    }

    /**
         * A single source query record with its server context (serverName, dbVendor).
         * Used to group queries by server when writing split output files.
         */
        private static class SourceQueryEntry {
        String serverName;
        String dbVendor;
        Map<String, Object> query;

        SourceQueryEntry(String serverName, String dbVendor, Map<String, Object> query) {
            this.serverName = serverName;
            this.dbVendor = dbVendor;
            this.query = query;
        }
    }
    
    /**
     * Calculate total size of source records
     */
    private static int calculateSourceSize(List<Map<?, ?>> records) {
        if (records == null) return 0;
        int size = 0;
        for (Map<?, ?> record : records) {
            Object sourceCode = record.get("sourceCode");
            if (sourceCode != null) {
                size += sourceCode.toString().length();
            }
        }
        return size;
    }
    
    /**
         * Create a catalog file for a batch of databases by converting sqlflow-sharded
         * catalog shards into the old sqlflow format.
         *
         * <p>The output uses the old sqlflow format ({@code servers[].databases[]},
         * no {@code format} key) so that the re-reader recognizes it via
         * {@link gudusoft.gsqlparser.dlineage.dataflow.metadata.MetadataReader#isSqlflow(String)}.
         * The {@code createdBy} must contain "sqlflow" for the re-reader to recognize
         * the format.
         *
         * @param outputDir  the directory to write the output file
         * @param inputFile  the original manifest file (used for naming)
         * @param extension  the file extension to use
         * @param batch      the batch of catalog shards to merge
         * @param batchIndex the batch index for file naming
         * @param createdBy  the createdBy string (must contain "sqlflow")
         * @return the created catalog file, or null if the batch is empty
         */
        private static File createCatalogBatchFile(File outputDir, File inputFile, String extension,
                                            List<DatabaseCatalogInfo> batch, int batchIndex,
                                            String createdBy) throws IOException {
        if (batch == null || batch.isEmpty()) {
            return null;
        }
        
        Map<String, Object> sqlflow = new LinkedHashMap<>();
        sqlflow.put("createdBy", createdBy != null ? createdBy : "sqlflow-ingester");
        
        // Add mergedFrom field to track original shardIds
        List<String> mergedFrom = new ArrayList<>();
        
        List<Map<String, Object>> servers = new ArrayList<>();
        
        for (DatabaseCatalogInfo catalog : batch) {
            mergedFrom.add(catalog.shardId);
            
            Map<String, Object> catalogContent = readCatalogContent(catalog.file);
            if (catalogContent == null) continue;
            
            Map<String, Object> serverObj = new LinkedHashMap<>();
            serverObj.put("name", catalog.serverName);
            serverObj.put("dbVendor", catalog.dbVendor);
            
            Map<String, Object> dbObj = new LinkedHashMap<>();
            dbObj.put("name", catalog.databaseName);
            
            copyCatalogFields(catalogContent, dbObj);
            
            List<Map<String, Object>> databases = new ArrayList<>();
            databases.add(dbObj);
            serverObj.put("databases", databases);
            servers.add(serverObj);
        }

        sqlflow.put("servers", servers);
        sqlflow.put("mergedFrom", mergedFrom);

        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_catalog_" + batchIndex +
                (extension.isEmpty() ? "" : "." + extension);
        
        File outputFile = new File(outputDir, fileName);
        writeJsonFile(outputFile, sqlflow);
        return outputFile;
    }
    
    /**
         * Create a catalog file for a single large database by converting its
         * sqlflow-sharded catalog shard into the old sqlflow format.
         *
         * <p>Same format as {@link #createCatalogBatchFile} but for a single
         * database whose catalog alone exceeds the split size threshold. The output
         * uses the old sqlflow format ({@code servers[].databases[]}, no
         * {@code format} key) so the re-reader recognizes it via
         * {@code MetadataReader.isSqlflow()}.
         *
         * @param outputDir    the directory to write the output file
         * @param inputFile    the original manifest file (used for naming)
         * @param catalog      the catalog shard to convert
         * @param extension    the file extension to use
         * @param catalogIndex the catalog index for file naming
         * @param createdBy    the createdBy string (must contain "sqlflow")
         * @return the created catalog file, or null on failure
         */
        private static File createCatalogSingleFile(File outputDir, File inputFile, DatabaseCatalogInfo catalog,
                                             String extension, int catalogIndex, String createdBy) throws IOException {
        Map<String, Object> catalogContent = readCatalogContent(catalog.file);
        if (catalogContent == null) {
            return null;
        }
        
        Map<String, Object> sqlflow = new LinkedHashMap<>();
        sqlflow.put("createdBy", createdBy != null ? createdBy : "sqlflow-ingester");
        
        Map<String, Object> serverObj = new LinkedHashMap<>();
        serverObj.put("name", catalog.serverName);
        serverObj.put("dbVendor", catalog.dbVendor);
        
        Map<String, Object> dbObj = new LinkedHashMap<>();
        dbObj.put("name", catalog.databaseName);
        copyCatalogFields(catalogContent, dbObj);
        
        List<Map<String, Object>> databases = new ArrayList<>();
        databases.add(dbObj);
        serverObj.put("databases", databases);
        
        List<Map<String, Object>> servers = new ArrayList<>();
        servers.add(serverObj);
        sqlflow.put("servers", servers);

        // Add mergedFrom field to track original shardId
        List<String> mergedFrom = new ArrayList<>();
        mergedFrom.add(catalog.shardId);
        sqlflow.put("mergedFrom", mergedFrom);
        
        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_catalog_" + catalogIndex +
                (extension.isEmpty() ? "" : "." + extension);
        
        File outputFile = new File(outputDir, fileName);
        writeJsonFile(outputFile, sqlflow);
        return outputFile;
    }
    
    /**
     * Read catalog content from file
     */
    private static Map<String, Object> readCatalogContent(File catalogFile) throws IOException {
        String content = readFileContent(catalogFile);
        @SuppressWarnings("unchecked")
        Map<String, Object> catalog = (Map<String, Object>) JSON.parseObject(content);
        return catalog;
    }
    
    /**
         * Copy catalog metadata fields from the raw catalog shard into the database
         * object of the output file. Preserves schemas, tables, views, procedures,
         * functions, triggers, packages, and synonyms.
         */
        private static void copyCatalogFields(Map<String, Object> catalog, Map<String, Object> dbObj) {
        if (catalog.containsKey("schemas")) dbObj.put("schemas", catalog.get("schemas"));
        if (catalog.containsKey("tables")) dbObj.put("tables", catalog.get("tables"));
        if (catalog.containsKey("views")) dbObj.put("views", catalog.get("views"));
        if (catalog.containsKey("procedures")) dbObj.put("procedures", catalog.get("procedures"));
        if (catalog.containsKey("functions")) dbObj.put("functions", catalog.get("functions"));
        if (catalog.containsKey("triggers")) dbObj.put("triggers", catalog.get("triggers"));
        if (catalog.containsKey("packages")) dbObj.put("packages", catalog.get("packages"));
        if (catalog.containsKey("synonyms")) dbObj.put("synonyms", catalog.get("synonyms"));
    }
    
    /**
         * Create a source query file for a batch of records by writing them in the
         * old sqlflow format.
         *
         * <p>Output format: {@code servers[].queries[]} with no {@code format} key.
         * Queries are grouped by server (serverName + dbVendor). The
         * {@code createdBy} is set from the caller-provided value and must contain
         * "sqlflow" for the re-reader to recognize the format via
         * {@code MetadataReader.isSqlflow()}.
         *
         * <p>File naming: {@code <inputName>_source_<startIndex>_<endIndex>.<ext>}
         *
         * @param outputDir  the directory to write the output file
         * @param inputFile  the original manifest file (used for naming)
         * @param extension  the file extension
         * @param entries    the source query records to write
         * @param mergedFrom the list of original shardIds
         * @param startIndex the start record index (for file naming)
         * @param endIndex   the end record index (for file naming)
         * @param createdBy  the createdBy string (must contain "sqlflow")
         * @return the created file
         */
        private static File createSourceQueryFileCommon(File outputDir, File inputFile, String extension,
                                                  List<SourceQueryEntry> entries, List<String> mergedFrom,
                                                  int startIndex, int endIndex,
                                                  String createdBy) throws IOException {
        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_source_" + startIndex + "_" + endIndex +
                (extension.isEmpty() ? "" : "." + extension);

        File queryFile = new File(outputDir, fileName);

        Map<String, Object> sqlflow = new LinkedHashMap<>();
        sqlflow.put("createdBy", createdBy != null ? createdBy : "grabit v1.7.0");
        sqlflow.put("mergedFrom", mergedFrom);

        // Group queries by server (preserve insertion order)
        Map<String, Map<String, Object>> serverMap = new LinkedHashMap<>();
        for (SourceQueryEntry entry : entries) {
            String key = (entry.serverName == null ? "" : entry.serverName) + "|" + (entry.dbVendor == null ? "" : entry.dbVendor);
            Map<String, Object> serverObj = serverMap.get(key);
            if (serverObj == null) {
                serverObj = new LinkedHashMap<>();
                serverObj.put("name", entry.serverName);
                if (entry.dbVendor != null) {
                    serverObj.put("dbVendor", entry.dbVendor);
                }
                serverObj.put("queries", new ArrayList<Map<String, Object>>());
                serverMap.put(key, serverObj);
            }
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> queries = (List<Map<String, Object>>) serverObj.get("queries");
            queries.add(entry.query);
        }

        List<Map<String, Object>> servers = new ArrayList<>(serverMap.values());
        sqlflow.put("servers", servers);

        writeJsonFile(queryFile, sqlflow);
        return queryFile;
    }
    
    /**
     * Resolve path relative to base directory
     */
    private static File resolvePath(File baseDir, String path) {
        if (path == null || path.isEmpty()) {
            return null;
        }
        
        File file = new File(path);
        if (file.isAbsolute()) {
            return file;
        }
        
        return new File(baseDir, path);
    }
    
    /**
         * Read source records from a JSONL file.
         *
         * <p>Supports two compression modes:
         * <ul>
         *   <li>{@code "block"} — gzip-compressed JSONL, read via GZIPInputStream.</li>
         *   <li>Otherwise — plain text JSONL, split by newline.</li>
         * </ul>
         *
         * @param sourceFile  the JSONL source file
         * @param compression the compression mode ("block" or null)
         * @return list of parsed records, or null if the file cannot be read
         */
        private static List<Map<?, ?>> readSourceRecords(File sourceFile, String compression) throws IOException {
        if (sourceFile == null || !sourceFile.exists()) {
            return null;
        }
        
        List<Map<?, ?>> records = new ArrayList<>();
        
        if ("block".equals(compression)) {
            // Read gzip compressed file
            try (java.util.zip.GZIPInputStream gzis = new java.util.zip.GZIPInputStream(
                    new java.io.FileInputStream(sourceFile))) {
                java.io.BufferedReader reader = new java.io.BufferedReader(
                        new java.io.InputStreamReader(gzis, "UTF-8"));
                String line;
                while ((line = reader.readLine()) != null) {
                    if (!line.trim().isEmpty()) {
                        Map<?, ?> record = (Map<?, ?>) JSON.parseObject(line);
                        records.add(record);
                    }
                }
            }
        } else {
            // Read plain JSONL file
            String content = readFileContent(sourceFile);
            String[] lines = content.split("\n");
            for (String line : lines) {
                if (!line.trim().isEmpty()) {
                    Map<?, ?> record = (Map<?, ?>) JSON.parseObject(line);
                    records.add(record);
                }
            }
        }
        
        return records;
    }
    
    /**
         * Holds metadata for a single catalog shard: server, database, and the
         * catalog file to read.
         */
        private static class DatabaseCatalogInfo {
        String serverName;
        String dbVendor;
        String databaseName;
        String shardId;
        File file;
        
        DatabaseCatalogInfo(String serverName, String dbVendor, String databaseName, String shardId, File file) {
            this.serverName = serverName;
            this.dbVendor = dbVendor;
            this.databaseName = databaseName;
            this.shardId = shardId;
            this.file = file;
        }
    }
    
    /**
         * Holds metadata for a single source shard: server, database, source file,
         * and compression mode.
         */
        private static class DatabaseSourceInfo {
        String serverName;
        String dbVendor;
        String databaseName;
        String shardId;
        File file;
        String compression;
        
        DatabaseSourceInfo(String serverName, String dbVendor, String databaseName, String shardId, File file, String compression) {
            this.serverName = serverName;
            this.dbVendor = dbVendor;
            this.databaseName = databaseName;
            this.shardId = shardId;
            this.file = file;
            this.compression = compression;
        }
    }

    /**
     * Create query file
     */
    private static File createQueryFile(File outputDir, File inputFile, long startIndex, long endIndex, String extension) {
        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_" + startIndex + "_" + endIndex + (extension.isEmpty() ? "" : "." + extension);
        return new File(outputDir, fileName);
    }

    private static Long getLastLineFromCurrentPos(RandomAccessFile raf, long currentPos, long fileLength, long splitSizeBytes, EDbVendor vendor) throws IOException {
        raf.seek(currentPos);

        long remainingBytes = fileLength - currentPos;
        long readSize = Math.min(splitSizeBytes, remainingBytes);

        byte[] buffer = new byte[(int) readSize];
        int bytesRead = raf.read(buffer);
        if (bytesRead <= 0) {
            return null;
        }

        String partialContent = new String(buffer, 0, bytesRead, "UTF-8");
        if (partialContent.isEmpty()) {
            return null;
        }

        TGSqlParser parser = new TGSqlParser(vendor);
        parser.sqltext = partialContent;
        parser.getrawsqlstatements();

        long lastValidatedLineNo = parser.getLastLineNoOfLastStatementBeenValidated();
        return lastValidatedLineNo;
    }

    /**
     * Find end position of Nth line from specified byte position
     * Supports multiple newline formats: \n (Unix/Linux), \r\n (Windows), \r (Old Mac)
     * Uses buffer for batch reading to improve performance
     */
    private static long findLineEndPosition(RandomAccessFile raf, long startPos, long lineCount) throws IOException {
        raf.seek(startPos);
        long newlineCount = 0;
        long fileLength = raf.length();

        byte[] buffer = new byte[BUFFER_SIZE];
        long currentPos = startPos;

        while (currentPos < fileLength) {
            int toRead = (int) Math.min(buffer.length, fileLength - currentPos);
            int bytesRead = raf.read(buffer, 0, toRead);
            if (bytesRead <= 0) {
                break;
            }

            for (int i = 0; i < bytesRead; i++) {
                byte b = buffer[i];

                if (b == '\n') {
                    newlineCount++;
                    if (newlineCount == lineCount) {
                        return currentPos + i + 1;
                    }
                } else if (b == '\r') {
                    if (i + 1 < bytesRead) {
                        if (buffer[i + 1] == '\n') {
                            newlineCount++;
                            if (newlineCount == lineCount) {
                                return currentPos + i + 2;
                            }
                            i++;
                        } else {
                            newlineCount++;
                            if (newlineCount == lineCount) {
                                return currentPos + i + 1;
                            }
                        }
                    } else {
                        long savedPos = raf.getFilePointer();
                        int nextByte = raf.read();

                        if (nextByte == '\n') {
                            newlineCount++;
                            if (newlineCount == lineCount) {
                                return currentPos + i + 2;
                            }
                            currentPos++;
                        } else {
                            newlineCount++;
                            if (newlineCount == lineCount) {
                                return currentPos + i + 1;
                            }
                            if (nextByte != -1) {
                                raf.seek(savedPos);
                            }
                        }
                    }
                }
            }

            currentPos += bytesRead;
        }

        return fileLength;
    }

    /**
     * Split file by byte position
     */
    private static File splitFileByPosition(File outputDir, File inputFile, long startPos, long endPos, int fileIndex, long startLineNo, long endLineNo) throws IOException {
        String fileName = getFileNameWithoutExtension(inputFile.getName()) + "_" + fileIndex + "_" + startLineNo + "_" + endLineNo + "." + getFileExtension(inputFile.getName());
        File outputFile = new File(outputDir, fileName);

        try (RandomAccessFile rafRead = new RandomAccessFile(inputFile, "r");
             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(outputFile), java.nio.charset.StandardCharsets.UTF_8))) {

            rafRead.seek(startPos);
            long bytesToRead = endPos - startPos;
            long bytesRemaining = bytesToRead;

            byte[] buffer = new byte[BUFFER_SIZE];

            while (bytesRemaining > 0) {
                int toRead = (int) Math.min(buffer.length, bytesRemaining);
                int bytesRead = rafRead.read(buffer, 0, toRead);
                if (bytesRead <= 0) {
                    break;
                }

                String chunk = new String(buffer, 0, bytesRead, java.nio.charset.StandardCharsets.UTF_8);
                writer.write(chunk);
                bytesRemaining -= bytesRead;
            }

            logger.info("split file " + inputFile.getName() + " (index: " + fileIndex + ", lines: " + startLineNo + "-" + endLineNo + ") from byte " + startPos + " to " + endPos + " to " + outputFile.getName());

            return outputFile;
        }
    }
}