[#220] fix: address CodeRabbit review findings

- create_table: 정리 경로를 FileSystem 트레이트로 추상화
  (remove_dir_all을 FileSystem에 추가, tokio::fs 직접 호출 제거)
- parser: PRIMARY KEY 목록의 쉼표 문법 검증
  ((,a), (a,), (a,,b), (a b) 거부 — 식별자/쉼표 교대 강제)
- parser: 테이블 레벨 PK 경로에도 문장 종료 검증 적용
  (PK 뒤 이상한 토큰 시 에러, 세미콜론 없는 EOF는 허용)
- e2e 테스트: 동기 Path::exists() 제거 (비동기 remove_dir_all로 정리)
- 신규 파서 테스트 3개 (쉼표 문법/후행 토큰/EOF 종료)

Author: DPS0340

src/common/fs.rs

@@ -20,6 +20,9 @@ pub trait FileSystem {
     /// 파일의 크기(bytes)를 반환합니다. (#265)
     /// `read_segment_rows`가 파일 전체를 메모리로 읽기 전에 예산을 확보하는 데 사용합니다.
     async fn metadata(&self, path: &Path) -> io::Result<u64>;
+    /// 디렉토리와 그 내용 전체를 재귀적으로 삭제합니다. (#220)
+    /// create_table 실패 시 생성 중인 테이블 디렉토리를 정리하는 데 사용합니다.
+    async fn remove_dir_all(&self, path: &Path) -> io::Result<()>;
 }
 
 pub struct RealFileSystem;
@@ -61,4 +64,8 @@ impl FileSystem for RealFileSystem {
         let metadata = tokio::fs::metadata(path).await?;
         Ok(metadata.len())
     }
+
+    async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
+        tokio::fs::remove_dir_all(path).await
+    }
 }

src/engine/actions/ddl/create_table.rs

@@ -76,7 +76,7 @@ impl DBEngine {
 
         if !primary_key_columns.is_empty() {
             if let Err(error) = self.ensure_indices_loaded().await {
-                let _ = tokio::fs::remove_dir_all(&table_path).await;
+                let _ = self.file_system.remove_dir_all(&table_path).await;
                 return Err(error);
             }
 
@@ -89,7 +89,7 @@ impl DBEngine {
             );
 
             if let Err(error) = self.index_manager.create_index(meta).await {
-                let _ = tokio::fs::remove_dir_all(&table_path).await;
+                let _ = self.file_system.remove_dir_all(&table_path).await;
                 return Err(error);
             }
         }

src/engine/actions/test_composite_pk_e2e.rs

@@ -24,9 +24,9 @@ use crate::engine::{DBEngine, SharedWALManager};
 
 async fn build_test_engine(test_name: &str) -> (DBEngine, SharedWALManager) {
     let base_path = PathBuf::from("target/test_composite_pk").join(test_name);
-    if base_path.exists() {
-        tokio::fs::remove_dir_all(&base_path).await.unwrap();
-    }
+
+    // 남은 이전 실행 정리 — remove_dir_all은 존재하지 않아도 실패를 신경 쓰지 않음
+    let _ = tokio::fs::remove_dir_all(&base_path).await;
 
     let config = LaunchConfig::default_for_base_path(&base_path);
     tokio::fs::create_dir_all(&config.data_directory)
@@ -139,7 +139,10 @@ async fn composite_pk_rejects_duplicate_key_combinations() {
     .await
     .expect("different combination with reused group_id must be accepted");
 
-    assert_eq!(engine.full_scan(memberships_table()).await.unwrap().len(), 3);
+    assert_eq!(
+        engine.full_scan(memberships_table()).await.unwrap().len(),
+        3
+    );
 }
 
 #[tokio::test]
@@ -213,7 +216,11 @@ async fn composite_pk_lookup_is_correct_on_a_large_table() {
     )
     .await
     .unwrap();
-    assert_eq!(result.rows.len(), 1, "user_id = 42 must match exactly one row");
+    assert_eq!(
+        result.rows.len(),
+        1,
+        "user_id = 42 must match exactly one row"
+    );
     assert_eq!(result.rows[0].fields[0], ExecuteField::Integer(42 % 7));
 
     // 존재하지 않는 키: 0행 (과거 버그로는 0행이 정답처럼 보였지만
@@ -296,7 +303,10 @@ async fn composite_pk_maintains_index_across_update_and_delete() {
         "update memberships set user_id = 2, group_id = 3 where user_id = 1 and group_id = 1;",
     )
     .await;
-    assert!(conflict.is_err(), "duplicate combination via update must fail");
+    assert!(
+        conflict.is_err(),
+        "duplicate combination via update must fail"
+    );
 
     // DELETE: 인덱스 항목 제거
     execute_sql(

src/engine/parser/implements/ddl/table.rs

@@ -90,6 +90,9 @@ impl Parser {
                     }
 
                     let mut columns: Vec<String> = vec![];
+                    // 식별자와 쉼표가 교대로 와야 합니다 (#220):
+                    // PRIMARY KEY (a, b) O, (,a) / (a,) / (a,,b) X
+                    let mut expect_identifier = true;
 
                     loop {
                         if !self.has_next_token() {
@@ -99,10 +102,32 @@ impl Parser {
                         let current_token = self.get_next_token();
 
                         match current_token {
-                            Token::RightParentheses => break,
-                            Token::Comma => continue,
+                            Token::RightParentheses => {
+                                if expect_identifier && !columns.is_empty() {
+                                    return Err(ParsingError::wrap(
+                                        "trailing comma in 'PRIMARY KEY (...)'".to_string(),
+                                    ));
+                                }
+                                break;
+                            }
+                            Token::Comma => {
+                                if expect_identifier {
+                                    return Err(ParsingError::wrap(
+                                        "expected column name in 'PRIMARY KEY (...)'. but your input word is 'Comma'"
+                                            .to_string(),
+                                    ));
+                                }
+                                expect_identifier = true;
+                            }
                             Token::Identifier(column_name) => {
+                                if !expect_identifier {
+                                    return Err(ParsingError::wrap(format!(
+                                        "expected ',' or ')' after column name in 'PRIMARY KEY (...)'. but your input word is '{:?}'",
+                                        column_name
+                                    )));
+                                }
                                 columns.push(column_name);
+                                expect_identifier = false;
                             }
                             _ => {
                                 return Err(ParsingError::wrap(format!(
@@ -172,6 +197,21 @@ impl Parser {
                 }
             }
 
+            // 문장 종료 검증은 일반 경로와 동일하게 적용합니다 (#220):
+            // 테이블 레벨 PK 뒤에 이상한 토큰이 남으면 에러.
+            if !self.has_next_token() {
+                return Ok(query);
+            }
+
+            let current_token = self.get_next_token();
+
+            if Token::SemiColon != current_token {
+                return Err(ParsingError::wrap(format!(
+                    "expected ';'. but your input word is '{:?}'",
+                    current_token
+                )));
+            }
+
             return Ok(query);
         }
 

src/engine/parser/test/create_table.rs

@@ -156,6 +156,61 @@ pub fn create_table_rejects_primary_key_referencing_unknown_column() {
     );
 }
 
+/// PK 목록의 문법 오류(쉼표 위치)를 거부해야 합니다 (#220, CodeRabbit)
+#[test]
+pub fn create_table_rejects_primary_key_comma_syntax_errors() {
+    let cases = [
+        "CREATE TABLE t (a INTEGER, PRIMARY KEY (,a));", // 선두 쉼표
+        "CREATE TABLE t (a INTEGER, PRIMARY KEY (a,));", // 후행 쉼표
+        "CREATE TABLE t (a INTEGER, PRIMARY KEY (a,,b));", // 연속 쉼표
+        "CREATE TABLE t (a INTEGER, PRIMARY KEY (a b));", // 쉼표 누락
+    ];
+
+    for text in cases {
+        let mut parser = Parser::with_string(text.to_owned()).unwrap();
+        assert!(
+            parser.parse(ParserContext::default()).is_err(),
+            "should reject: {}",
+            text
+        );
+    }
+}
+
+/// 테이블 레벨 PK 뒤에 이상한 토큰이 남으면 에러 (#220, CodeRabbit)
+#[test]
+pub fn create_table_rejects_trailing_tokens_after_table_level_primary_key() {
+    let text = r#"
+        CREATE TABLE "test_db".t
+        (
+            a INTEGER,
+            PRIMARY KEY (a)
+        ) unexpected;
+    "#
+    .to_owned();
+
+    let mut parser = Parser::with_string(text).unwrap();
+
+    assert!(parser.parse(ParserContext::default()).is_err());
+}
+
+/// 테이블 레벨 PK + 세미콜론 없는 종료(EOF)는 정상 파싱 (#220)
+#[test]
+pub fn create_table_accepts_table_level_primary_key_without_semicolon() {
+    let text = r#"
+        CREATE TABLE "test_db".t
+        (
+            a INTEGER,
+            PRIMARY KEY (a)
+        )
+    "#
+    .to_owned();
+
+    let mut parser = Parser::with_string(text).unwrap();
+
+    let statements = parser.parse(ParserContext::default()).unwrap();
+    assert_eq!(statements.len(), 1);
+}
+
 #[test]
 pub fn create_table() {
     let text = r#"