csv_transform_in_mem_with_loom.zig

// CSV Transform - Parallel CSV Processing Demo (Parallel File Read Version)
//
// Demonstrates parallel CSV processing using loom.
// Uses loom to read file chunks in parallel, then processes with parallel iterators.
//
// Usage: zig build samples-loom -Doptimize=ReleaseFast
//
// Input:  src/loom/docs/datas/inputs/sample_01.csv (auto-generated if missing)
// Output: Aggregation results and statistics

const std = @import("std");
const loom = @import("loom");
const par_iter = loom.par_iter;
const par_range = loom.par_range;
const ThreadPool = loom.ThreadPool;

const input_file = "src/loom/docs/datas/inputs/sample_01.csv";

// Default row count for generated CSV (100K for quick demos, increase for stress testing)
const DEFAULT_ROW_COUNT: usize = 100_000;

// CSV generation categories and descriptions (matches random_csv_gen.py)
const categories = [_][]const u8{ "Apple", "Banana", "Cherry", "Date", "Elderberry", "Fig", "Grape", "Honeydew" };
const descriptions = [_][]const u8{
    "lorem ipsum dolor",
    "sit amet consectetur",
    "adipiscing elit",
    "sed do eiusmod",
    "tempor incididunt",
    "ut labore et dolore",
    "magna aliqua",
    "ut enim ad minim",
};

/// Generates sample CSV file if it doesn't exist.
/// Mirrors the behavior of src/loom/docs/datas/random_csv_gen.py
fn ensureSampleCsvExists(allocator: std.mem.Allocator, total_rows: usize) !void {
    // Check if file already exists
    std.fs.cwd().access(input_file, .{}) catch {
        // File doesn't exist, generate it
        std.debug.print("Sample CSV not found, generating {d} rows...\n", .{total_rows});
        try generateSampleCsv(allocator, total_rows);
        return;
    };
    std.debug.print("Sample CSV exists: {s}\n", .{input_file});
}

/// Generates a sample CSV file with random data
fn generateSampleCsv(allocator: std.mem.Allocator, total_rows: usize) !void {
    var timer = try std.time.Timer.start();

    // Ensure parent directory exists
    const dir_path = std.fs.path.dirname(input_file) orelse ".";
    std.fs.cwd().makePath(dir_path) catch {};

    const file = try std.fs.cwd().createFile(input_file, .{});
    defer file.close();

    // Write header
    try file.writeAll("id,measurement,category,description,is_active\n");

    // Initialize PRNG
    var prng = std.Random.DefaultPrng.init(@bitCast(std.time.timestamp()));
    const random = prng.random();

    // Generate rows in chunks, building each chunk in memory before writing
    var rows_written: usize = 0;
    const chunk_size: usize = 10_000;

    // Buffer for building chunks
    var chunk_buf: std.ArrayListUnmanaged(u8) = .empty;
    defer chunk_buf.deinit(allocator);

    while (rows_written < total_rows) {
        const chunk_end = @min(rows_written + chunk_size, total_rows);

        // Clear buffer for new chunk
        chunk_buf.shrinkRetainingCapacity(0);

        for (rows_written..chunk_end) |id| {
            // Random measurement 0.0-1000.0 with 4 decimal places
            const measurement = random.float(f64) * 1000.0;

            // Random category and description
            const category = categories[random.intRangeAtMost(usize, 0, categories.len - 1)];
            const description = descriptions[random.intRangeAtMost(usize, 0, descriptions.len - 1)];

            // Random is_active (0 or 1)
            const is_active: u8 = random.intRangeAtMost(u8, 0, 1);

            // Build row string
            const row = try std.fmt.allocPrint(allocator, "{d},{d:.4},{s},{s},{d}\n", .{ id, measurement, category, description, is_active });
            defer allocator.free(row);
            try chunk_buf.appendSlice(allocator, row);
        }

        // Write entire chunk at once
        try file.writeAll(chunk_buf.items);

        rows_written = chunk_end;

        // Progress update
        const elapsed_ns = timer.read();
        const elapsed_s = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000_000.0;
        std.debug.print("  Written {d} rows... ({d:.2}s elapsed)\n", .{ rows_written, elapsed_s });
    }

    const total_elapsed = timer.read();
    const file_stat = try file.stat();

    std.debug.print("Generated {s}\n", .{input_file});
    std.debug.print("  Rows: {d}, Size: {d:.2} MB, Time: {d:.2}s\n\n", .{
        total_rows,
        @as(f64, @floatFromInt(file_stat.size)) / (1024.0 * 1024.0),
        @as(f64, @floatFromInt(total_elapsed)) / 1_000_000_000.0,
    });
}

// Chunk size for parallel file reading (256MB per chunk - larger = fewer syscalls)
const READ_CHUNK_SIZE: usize = 256 * 1024 * 1024;

pub fn main() !void {
    const allocator = std.heap.page_allocator;

    std.debug.print("\n", .{});
    std.debug.print("================================================\n", .{});
    std.debug.print("  CSV Transform - Parallel File Read with Loom\n", .{});
    std.debug.print("================================================\n\n", .{});

    // Ensure sample CSV exists (generates if missing)
    try ensureSampleCsvExists(allocator, DEFAULT_ROW_COUNT);

    // Initialize thread pool FIRST (before file operations)
    const pool = try ThreadPool.init(allocator, .{});
    defer pool.deinit();
    std.debug.print("Thread pool: {d} workers\n\n", .{pool.numWorkers()});

    // ========================================================================
    // PARALLEL FILE READING using loom
    // ========================================================================
    std.debug.print("--- Parallel File Reading (using loom) ---\n\n", .{});
    std.debug.print("Opening: {s}\n", .{input_file});

    var read_timer = try std.time.Timer.start();

    const file = std.fs.cwd().openFile(input_file, .{}) catch |err| {
        std.debug.print("Failed to open file: {any}\n", .{err});
        return err;
    };
    defer file.close();

    // Get file size
    const file_stat = try file.stat();
    const file_size = file_stat.size;

    std.debug.print("File size: {d:.2} GB ({d} bytes)\n", .{
        @as(f64, @floatFromInt(file_size)) / (1024.0 * 1024.0 * 1024.0),
        file_size,
    });

    // Allocate buffer for entire file
    const file_data = try allocator.alloc(u8, file_size);
    defer allocator.free(file_data);

    // Calculate number of chunks for parallel reading
    const num_chunks = (file_size + READ_CHUNK_SIZE - 1) / READ_CHUNK_SIZE;
    std.debug.print("Reading in {d} parallel chunks of {d}MB each...\n", .{ num_chunks, READ_CHUNK_SIZE / (1024 * 1024) });

    // Context for parallel file reading - shared across all workers
    const ReadContext = struct {
        file_handle: std.fs.File.Handle,
        buffer: []u8,
        chunk_size: usize,
        total_size: usize,
    };

    const read_ctx = ReadContext{
        .file_handle = file.handle,
        .buffer = file_data,
        .chunk_size = READ_CHUNK_SIZE,
        .total_size = file_size,
    };

    // Use par_range with context - clean API, no allocations needed!
    par_range(@as(usize, 0), num_chunks)
        .withPool(pool)
        .withContext(&read_ctx)
        .forEach(struct {
        fn readChunk(ctx: *const ReadContext, chunk_idx: usize) void {
            const start = chunk_idx * ctx.chunk_size;
            const end = @min(start + ctx.chunk_size, ctx.total_size);
            if (start >= end) return;
            // Use pread for positional read (thread-safe, doesn't move file cursor)
            _ = std.posix.pread(ctx.file_handle, ctx.buffer[start..end], start) catch {};
        }
    }.readChunk);

    const total_bytes_read = file_size;

    const read_elapsed = read_timer.read();
    const read_throughput = @as(f64, @floatFromInt(file_size)) / (@as(f64, @floatFromInt(read_elapsed)) / 1_000_000_000.0) / (1024.0 * 1024.0 * 1024.0);

    std.debug.print("Parallel read: {d:.2}s ({d:.2} GB/s)\n", .{
        @as(f64, @floatFromInt(read_elapsed)) / 1_000_000_000.0,
        read_throughput,
    });
    std.debug.print("Total bytes: {d}\n\n", .{total_bytes_read});

    // ========================================================================
    // PARALLEL FILE ANALYSIS - Process in-memory data with loom
    // ========================================================================
    std.debug.print("--- Parallel File Analysis (using loom) ---\n\n", .{});

    var total_timer = try std.time.Timer.start();

    // Count newlines in parallel
    const newline_count = par_iter(file_data).withPool(pool).count(struct {
        fn isNewline(byte: u8) bool {
            return byte == '\n';
        }
    }.isNewline);

    var elapsed = total_timer.read();
    const throughput_1 = @as(f64, @floatFromInt(file_size)) / (@as(f64, @floatFromInt(elapsed)) / 1_000_000_000.0) / (1024.0 * 1024.0 * 1024.0);
    std.debug.print("Newline count (parallel): {d:.2}ms ({d:.2} GB/s) -> {d} lines\n", .{
        @as(f64, @floatFromInt(elapsed)) / 1_000_000.0,
        throughput_1,
        newline_count,
    });

    // Count commas in parallel
    total_timer.reset();
    const comma_count = par_iter(file_data).withPool(pool).count(struct {
        fn isComma(byte: u8) bool {
            return byte == ',';
        }
    }.isComma);
    elapsed = total_timer.read();
    const throughput_2 = @as(f64, @floatFromInt(file_size)) / (@as(f64, @floatFromInt(elapsed)) / 1_000_000_000.0) / (1024.0 * 1024.0 * 1024.0);
    const avg_fields = @as(f64, @floatFromInt(comma_count)) / @as(f64, @floatFromInt(newline_count));
    std.debug.print("Comma count (parallel):   {d:.2}ms ({d:.2} GB/s) -> {d} commas ({d:.1} fields/row)\n", .{
        @as(f64, @floatFromInt(elapsed)) / 1_000_000.0,
        throughput_2,
        comma_count,
        avg_fields + 1,
    });

    // Count digit characters in parallel
    total_timer.reset();
    const digit_count = par_iter(file_data).withPool(pool).count(struct {
        fn isDigit(byte: u8) bool {
            return byte >= '0' and byte <= '9';
        }
    }.isDigit);
    elapsed = total_timer.read();
    const throughput_3 = @as(f64, @floatFromInt(file_size)) / (@as(f64, @floatFromInt(elapsed)) / 1_000_000_000.0) / (1024.0 * 1024.0 * 1024.0);
    const digit_pct = @as(f64, @floatFromInt(digit_count)) * 100.0 / @as(f64, @floatFromInt(file_size));
    std.debug.print("Digit count (parallel):   {d:.2}ms ({d:.2} GB/s) -> {d} ({d:.1}%% of file)\n\n", .{
        @as(f64, @floatFromInt(elapsed)) / 1_000_000.0,
        throughput_3,
        digit_count,
        digit_pct,
    });

    // ========================================================================
    // Parallel Line Indexing
    // ========================================================================
    std.debug.print("--- Parallel Line Indexing ---\n\n", .{});

    total_timer.reset();

    var lines = std.ArrayListUnmanaged([]const u8){};
    defer lines.deinit(allocator);
    try lines.ensureTotalCapacity(allocator, newline_count);

    var line_iter = std.mem.splitScalar(u8, file_data, '\n');
    _ = line_iter.next(); // Skip header

    while (line_iter.next()) |line| {
        if (line.len > 0) {
            lines.appendAssumeCapacity(line);
        }
    }

    const split_elapsed = total_timer.read();
    const row_count = lines.items.len;
    std.debug.print("Lines indexed: {d:.2}s ({d} rows)\n\n", .{
        @as(f64, @floatFromInt(split_elapsed)) / 1_000_000_000.0,
        row_count,
    });

    // ========================================================================
    // Parallel Aggregations
    // ========================================================================
    std.debug.print("--- Parallel Row Aggregations ---\n\n", .{});

    // 1. Count active rows
    total_timer.reset();
    const active_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            if (line.len == 0) return false;
            return line[line.len - 1] == '1';
        }
    }.check);
    elapsed = total_timer.read();
    const active_pct = @as(f64, @floatFromInt(active_count)) * 100.0 / @as(f64, @floatFromInt(row_count));
    std.debug.print("1. Active count:     {d:.2}ms\n", .{@as(f64, @floatFromInt(elapsed)) / 1_000_000.0});
    std.debug.print("   Result: {d} / {d} ({d:.1}%)\n\n", .{ active_count, row_count, active_pct });

    // 2. Count by category
    total_timer.reset();
    const apple_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Apple,") != null;
        }
    }.check);
    const banana_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Banana,") != null;
        }
    }.check);
    const cherry_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Cherry,") != null;
        }
    }.check);
    const date_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Date,") != null;
        }
    }.check);
    const fig_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Fig,") != null;
        }
    }.check);
    const grape_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Grape,") != null;
        }
    }.check);
    const honeydew_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, ",Honeydew,") != null;
        }
    }.check);
    elapsed = total_timer.read();
    std.debug.print("2. Category counts:  {d:.2}ms\n", .{@as(f64, @floatFromInt(elapsed)) / 1_000_000.0});
    std.debug.print("   Apple:    {d:>12}\n", .{apple_count});
    std.debug.print("   Banana:   {d:>12}\n", .{banana_count});
    std.debug.print("   Cherry:   {d:>12}\n", .{cherry_count});
    std.debug.print("   Date:     {d:>12}\n", .{date_count});
    std.debug.print("   Fig:      {d:>12}\n", .{fig_count});
    std.debug.print("   Grape:    {d:>12}\n", .{grape_count});
    std.debug.print("   Honeydew: {d:>12}\n\n", .{honeydew_count});

    // 3. Count high measurements (>900)
    total_timer.reset();
    const high_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            var field_iter = std.mem.splitScalar(u8, line, ',');
            _ = field_iter.next();
            const measurement_str = field_iter.next() orelse return false;
            const value = std.fmt.parseFloat(f64, measurement_str) catch return false;
            return value > 900.0;
        }
    }.check);
    elapsed = total_timer.read();
    const high_pct = @as(f64, @floatFromInt(high_count)) * 100.0 / @as(f64, @floatFromInt(row_count));
    std.debug.print("3. High values (>900): {d:.2}ms\n", .{@as(f64, @floatFromInt(elapsed)) / 1_000_000.0});
    std.debug.print("   Result: {d} ({d:.2}%)\n\n", .{ high_count, high_pct });

    // 4. Count low measurements (<100)
    total_timer.reset();
    const low_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            var field_iter = std.mem.splitScalar(u8, line, ',');
            _ = field_iter.next();
            const measurement_str = field_iter.next() orelse return false;
            const value = std.fmt.parseFloat(f64, measurement_str) catch return false;
            return value < 100.0;
        }
    }.check);
    elapsed = total_timer.read();
    const low_pct = @as(f64, @floatFromInt(low_count)) * 100.0 / @as(f64, @floatFromInt(row_count));
    std.debug.print("4. Low values (<100): {d:.2}ms\n", .{@as(f64, @floatFromInt(elapsed)) / 1_000_000.0});
    std.debug.print("   Result: {d} ({d:.2}%)\n\n", .{ low_count, low_pct });

    // 5. Count lorem ipsum
    total_timer.reset();
    const lorem_count = par_iter(lines.items).withPool(pool).count(struct {
        fn check(line: []const u8) bool {
            return std.mem.indexOf(u8, line, "lorem ipsum") != null;
        }
    }.check);
    elapsed = total_timer.read();
    const lorem_pct = @as(f64, @floatFromInt(lorem_count)) * 100.0 / @as(f64, @floatFromInt(row_count));
    std.debug.print("5. Contains 'lorem ipsum': {d:.2}ms\n", .{@as(f64, @floatFromInt(elapsed)) / 1_000_000.0});
    std.debug.print("   Result: {d} ({d:.2}%)\n\n", .{ lorem_count, lorem_pct });

    // ========================================================================
    // Summary
    // ========================================================================
    const avg_throughput = (throughput_1 + throughput_2 + throughput_3) / 3.0;
    std.debug.print("================================================\n", .{});
    std.debug.print("   Processing Complete!\n", .{});
    std.debug.print("================================================\n\n", .{});
    std.debug.print("Processed {d} rows with {d} workers\n", .{ row_count, pool.numWorkers() });
    std.debug.print("File size: {d:.2} GB\n", .{@as(f64, @floatFromInt(file_size)) / (1024.0 * 1024.0 * 1024.0)});
    std.debug.print("Parallel file read: {d:.2} GB/s\n", .{read_throughput});
    std.debug.print("Processing throughput: {d:.2} GB/s\n\n", .{avg_throughput});
}