Add CI job for integration tests and fix concurrent test

- Add separate CI job 'integration-test' that runs ignored tests with 15min timeout
- Add Makefile targets: test-integration and test-integration-minio
- Fix test_concurrent_mixed_operations to test concurrent writes to different
  projects (avoids delta conflict retries) and concurrent reads

Author: tonyalaribe

.github/workflows/ci.yml

@@ -94,3 +94,58 @@ jobs:
 
       - name: Run tests
         run: cargo test --all-features
+
+  integration-test:
+    name: Integration Tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    env:
+      AWS_SDK_LOAD_CONFIG: "false"
+      AWS_ENDPOINT_URL: http://127.0.0.1:9000
+      AWS_REGION: us-east-1
+      AWS_S3_BUCKET: timefusion-test
+      AWS_S3_ENDPOINT: http://127.0.0.1:9000
+      AWS_ALLOW_HTTP: "true"
+      AWS_ACCESS_KEY_ID: minioadmin
+      AWS_SECRET_ACCESS_KEY: minioadmin
+      PGWIRE_PORT: "12345"
+      PORT: "8080"
+      TIMEFUSION_TABLE_PREFIX: timefusion-ci-integration
+      BATCH_INTERVAL_MS: "1000"
+      MAX_BATCH_SIZE: "1000"
+      ENABLE_BATCH_QUEUE: "true"
+      MAX_PG_CONNECTIONS: "100"
+      AWS_S3_LOCKING_PROVIDER: ""
+      TIMEFUSION_FOYER_MEMORY_MB: "256"
+      TIMEFUSION_FOYER_DISK_GB: "10"
+      TIMEFUSION_FOYER_TTL_SECONDS: "300"
+      TIMEFUSION_FOYER_SHARDS: "8"
+    steps:
+      - name: Free disk space
+        run: sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc /opt/hostedtoolcache/CodeQL
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+
+      - name: Start MinIO
+        run: |
+          docker run -d -p 9000:9000 --name minio \
+            -e MINIO_ROOT_USER=minioadmin \
+            -e MINIO_ROOT_PASSWORD=minioadmin \
+            minio/minio server /data
+          sleep 5
+          until curl -sf http://localhost:9000/minio/health/live; do sleep 1; done
+
+      - name: Install AWS CLI
+        run: |
+          curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
+          unzip -q awscliv2.zip
+          sudo ./aws/install --update
+
+      - name: Create MinIO bucket
+        run: |
+          aws --endpoint-url http://127.0.0.1:9000 s3 mb s3://timefusion-test || true
+          aws --endpoint-url http://127.0.0.1:9000 s3 mb s3://timefusion-tests || true
+
+      - name: Run integration tests
+        run: cargo test --test integration_test --test sqllogictest -- --ignored

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test test-ovh test-minio test-prod run-prod build-prod minio-start minio-stop minio-clean
+.PHONY: test test-ovh test-minio test-prod test-integration test-integration-minio run-prod build-prod minio-start minio-stop minio-clean
 
 # Default test with MinIO/test environment (uses .env)
 test:
@@ -50,4 +50,15 @@ minio-stop:
 # Clean MinIO data
 minio-clean:
 	@rm -rf /tmp/minio-data
-	@echo "MinIO data cleaned"
+	@echo "MinIO data cleaned"
+
+# Run integration tests (postgres wire protocol tests, sqllogictests)
+# These are slower tests that start a full PGWire server
+test-integration:
+	@echo "Running integration tests..."
+	@export $$(cat .env | grep -v '^#' | xargs) && cargo test --test integration_test --test sqllogictest -- --ignored $${ARGS}
+
+# Run integration tests with MinIO
+test-integration-minio:
+	@echo "Running integration tests with MinIO..."
+	@export $$(cat .env.minio | grep -v '^#' | xargs) && cargo test --test integration_test --test sqllogictest -- --ignored $${ARGS}

src/database.rs

@@ -2372,18 +2372,43 @@ mod tests {
             let db = Database::new().await?;
             let db = Arc::new(db);
 
-            let project_id = format!("mixed_ops_{}", uuid::Uuid::new_v4());
+            // Test concurrent writes to DIFFERENT projects (no conflicts)
+            let mut handles = Vec::new();
+            for i in 0..3 {
+                let db_clone = Arc::clone(&db);
+                let project_id = format!("project_{}", i);
+                handles.push(tokio::spawn(async move {
+                    let batch = json_to_batch(vec![test_span(
+                        &format!("id_{}", i),
+                        &format!("span_{}", i),
+                        &project_id,
+                    )])?;
+                    db_clone.insert_records_batch(&project_id, "otel_logs_and_spans", vec![batch], true).await?;
+                    Ok::<_, anyhow::Error>(())
+                }));
+            }
 
-            // Sequential writes first, then optimize (reduced concurrency to speed up test)
+            // Wait for all writes
+            for handle in handles {
+                handle.await??;
+            }
+
+            // Now test concurrent reads across all projects
+            let mut read_handles = Vec::new();
             for i in 0..3 {
-                let batch_id = format!("batch_{}", i);
-                let batch = json_to_batch(vec![test_span(&batch_id, &format!("test_{}", batch_id), &project_id)])?;
-                db.insert_records_batch(&project_id, "otel_logs_and_spans", vec![batch], true).await?;
+                let db_clone = Arc::clone(&db);
+                let project_id = format!("project_{}", i);
+                read_handles.push(tokio::spawn(async move {
+                    let ctx = db_clone.clone().create_session_context();
+                    let _ = ctx.sql(&format!(
+                        "SELECT COUNT(*) FROM otel_logs_and_spans WHERE project_id = '{}'", project_id
+                    )).await;
+                    Ok::<_, anyhow::Error>(())
+                }));
             }
 
-            // Run optimize after writes
-            if let Ok(table_ref) = db.get_or_create_table(&project_id, "otel_logs_and_spans").await {
-                let _ = db.optimize_table(&table_ref, "otel_logs_and_spans", Some(1024 * 1024)).await;
+            for handle in read_handles {
+                handle.await??;
             }
 
             db.shutdown().await?;