Rust code that will trigger the errors

/// Test Issue #1: Race Condition - Query while file is being written
/// This simulates the scenario where a query executes while an insert is still writing to a parquet file
#[allow(dead_code)]
async fn test_race_condition_insert_query() -> Result<(), Box<dyn std::error::Error>> {
  println!("\n=== TESTING RACE CONDITION (Issue #1) ===");
  println!("This test triggers a query while an insert is writing to a parquet file");

  const STORAGE_PATH: &str = "tmp";
  const USERNAME: &str = "ahmed_test";
  const DATABASE_NAME: &str = "test_race_condition";
  const TABLE_NAME: &str = "race_test_table";

  let _ = init_timon(STORAGE_PATH, 1440, USERNAME).unwrap(); // Daily partitions
  let _ = create_database(DATABASE_NAME);

  let table_schema = r#"
    {
      "date": {
        "type": "int",
        "required": true,
        "unique": true,
        "datetime": true
      },
      "value": {
        "type": "int"
      }
    }
  "#;

  let _ = create_table(DATABASE_NAME, TABLE_NAME, &table_schema);
  println!("✅ Table created");

  // First, insert some initial data to ensure table is registered
  println!("📝 Inserting initial data to register table...");
  let initial_data = format!(
    r#"[{{"date":"{}","value":999}}]"#,
    (Utc::now() - chrono::Duration::days(2)).format("%Y.%m.%d %H:%M:%S")
  );
  let _ = insert(DATABASE_NAME, TABLE_NAME, &initial_data)?;
  println!("✅ Initial data inserted");

  // Query once to register the table
  let _ = query(DATABASE_NAME, &format!("SELECT COUNT(*) as count FROM {}", TABLE_NAME), None, None).await?;
  println!("✅ Table registered with DataFusion");

  // Now generate a VERY large dataset targeting the SAME partition as initial data
  // This will trigger an UPDATE operation which reads existing file, modifies it, and writes back
  // Using a much larger dataset to ensure the write takes significant time
  let target_date = Utc::now().format("%Y.%m.%d %H:%M:%S").to_string();
  let mut json_data = String::from("[");
  for i in 0..20000 {
    // Use same date to target same partition, but different unique values
    let date_str = target_date.clone();
    if i > 0 {
      json_data.push_str(",");
    }
    json_data.push_str(&format!(r#"{{"date":"{}","value":{}}}"#, date_str, i));
  }
  json_data.push_str("]");

  println!("🚀 Starting VERY large insert/update (20000 records - this will take significant time)...");

  // Start insert in a separate thread (not async task) to ensure it blocks
  let db_name = DATABASE_NAME.to_string();
  let table_name = TABLE_NAME.to_string();
  let data = json_data.clone();

  let insert_handle = std::thread::spawn(move || {
    println!("   Insert thread: Starting write operation (this will take time)...");
    let start = std::time::Instant::now();
    let result = insert(&db_name, &table_name, &data);
    let duration = start.elapsed();
    match result {
      Ok(_) => println!("   Insert thread: ✅ Write completed in {:.2}s", duration.as_secs_f64()),
      Err(e) => println!("   Insert thread: ❌ Write failed: {}", e),
    }
  });

  // Wait a bit for insert to start writing (file should be open and being written)
  // Reduced delay to catch it earlier in the write process
  std::thread::sleep(std::time::Duration::from_millis(10));

  // Now query while file is being written - do this MULTIPLE times to increase chance
  println!("⚡ Starting query WHILE insert is writing (race condition)...");
  println!("   Will query multiple times to increase chance of catching the error...");

  let mut error_found = false;
  for query_attempt in 1..=10 {
    std::thread::sleep(std::time::Duration::from_millis(20));
    println!("   Query attempt #{}...", query_attempt);

    let query_result = query(DATABASE_NAME, &format!("SELECT COUNT(*) as count FROM {}", TABLE_NAME), None, None).await;

    match query_result {
      Ok(result) => {
        // Check if the result indicates an error (status != 200 or error message)
        let status = result.get("status").and_then(|s| s.as_u64()).unwrap_or(0);
        let message = result.get("message").and_then(|m| m.as_str()).unwrap_or("");

        // Check for the EXACT error: "Invalid partitioning found on disk"
        if message.contains("Invalid partitioning found on disk") {
          println!("🎯 SUCCESS! Race condition triggered the EXACT error:");
          println!("   Status: {}, Message: {}", status, message);
          println!("   This confirms Issue #1: Race condition causes 'Invalid partitioning found on disk'");
          error_found = true;
          break;
        } else if status != 200
          || message.contains("Invalid partitioning")
          || message.contains("QueryExecutionFailed")
          || message.contains("Execution error")
          || message.contains("Out of range of file")
        {
          println!("⚠️  Related error found (but not exact 'Invalid partitioning found on disk'):");
          println!("   Status: {}, Message: {}", status, message);
          println!("   This is a related race condition error, but we need to find the exact one");
        } else if query_attempt == 1 {
          println!("   Attempt #{}: Query succeeded, continuing...", query_attempt);
        }
      }
      Err(e) => {
        let error_str = e.to_string();
        // Check for the EXACT error: "Invalid partitioning found on disk"
        if error_str.contains("Invalid partitioning found on disk") {
          println!("🎯 SUCCESS! Race condition triggered the EXACT error:");
          println!("   Error: {}", error_str);
          println!("   This confirms Issue #1: Race condition causes 'Invalid partitioning found on disk'");
          error_found = true;
          break;
        } else if error_str.contains("Invalid partitioning")
          || error_str.contains("partition")
          || error_str.contains("corrupted")
          || error_str.contains("Out of range of file")
        {
          println!("⚠️  Related error found (but not exact 'Invalid partitioning found on disk'):");
          println!("   Error: {}", error_str);
          println!("   This is a related race condition error, but we need to find the exact one");
        } else {
          println!("   Attempt #{}: Different error: {}", query_attempt, error_str);
        }
      }
    }
  }

  if !error_found {
    println!("⚠️  No error detected after 10 query attempts");
    println!("   The insert may have finished too quickly, or file locking prevented the issue");
  }

  // Wait for insert to complete
  let _ = insert_handle.join();

  Ok(())
}

Bash script that will keep calling the Rust code

#!/bin/bash

# Script to validate that the race condition fix works
# This test should find NO errors after the fix is implemented
# Usage: ./test_race_condition_validation.sh [max_iterations]

cd /home/ahmed/mongrov/rn-timon/timon

# Set environment variable to run test #1
export TIMON_TEST_ISSUE=1

ITERATION=0
MAX_ITERATIONS=${1:-100}
ERRORS_FOUND=0

echo "=========================================="
echo "Race Condition Fix Validation Test"
echo "This test validates that the fix prevents race condition errors"
echo "Max iterations: $MAX_ITERATIONS"
echo "Press Ctrl+C to stop"
echo "=========================================="
echo ""

while [ $ITERATION -lt $MAX_ITERATIONS ]; do
  ITERATION=$((ITERATION + 1))
  
  echo ""
  echo "=========================================="
  echo "Iteration #$ITERATION"
  echo "Time: $(date)"
  echo "=========================================="
  
  OUTPUT=$(timeout 120 env TIMON_TEST_ISSUE=1 cargo run 2>&1)
  EXIT_CODE=$?
  
  # Check for ANY race condition errors (should be NONE after fix)
  if echo "$OUTPUT" | grep -q "Out of range of file.*expected.*actual: 0\|FileReadFailed.*Out of range\|Invalid partitioning found on disk"; then
    ERRORS_FOUND=$((ERRORS_FOUND + 1))
    echo ""
    echo "=========================================="
    echo "❌ ERROR FOUND on iteration #$ITERATION!"
    echo "Time: $(date)"
    echo "=========================================="
    echo "$OUTPUT" | grep -i "Out of range\|Invalid partitioning\|FileReadFailed" | head -5
    echo ""
    echo "$OUTPUT" > "/tmp/timon_validation_error_iteration_${ITERATION}.log"
    echo "Full output saved to: /tmp/timon_validation_error_iteration_${ITERATION}.log"
    echo ""
    echo "❌ FAILURE: Fix is not working - error found!"
    echo "Stopping validation test."
    break
  else
    echo "✅ No errors found in iteration #$ITERATION"
  fi
  
  if [ $EXIT_CODE -eq 124 ]; then
    echo "⚠️  Test timed out after 120 seconds"
  elif [ $EXIT_CODE -ne 0 ]; then
    echo "⚠️  Test exited with code: $EXIT_CODE"
  fi
  
  sleep 1
done

echo ""
echo "=========================================="
echo "Validation Test Complete"
echo "Iterations Run: $ITERATION"
echo "Errors Found: $ERRORS_FOUND"
echo "=========================================="

if [ $ERRORS_FOUND -eq 0 ]; then
  echo "✅ SUCCESS: No race condition errors found after $ITERATION iterations - fix appears to be working!"
  exit 0
else
  echo "❌ FAILURE: Found $ERRORS_FOUND error(s) on iteration #$ITERATION - fix needs adjustment"
  exit 1
fi

bug: fix insert-query race condition preventing queries from reading incomplete parquet files #135

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions