[#220] feat: auto-create composite primary-key indexes

src/common/fs.rs

@@ -20,6 +20,14 @@ pub trait FileSystem {
     /// 파일의 크기(bytes)를 반환합니다. (#265)
     /// `read_segment_rows`가 파일 전체를 메모리로 읽기 전에 예산을 확보하는 데 사용합니다.
     async fn metadata(&self, path: &Path) -> io::Result<u64>;
+    /// 디렉토리와 그 내용 전체를 재귀적으로 삭제합니다. (#220)
+    /// create_table 실패 시 생성 중인 테이블 디렉토리를 정리하는 데 사용합니다.
+    ///
+    /// 트레이트 하위 호환을 위해 default 구현을 제공합니다 — 기존 외부
+    /// 구현체는 이 메서드 없이도 컴파일이 유지됩니다.
+    async fn remove_dir_all(&self, path: &Path) -> io::Result<()> {
+        tokio::fs::remove_dir_all(path).await
+    }
 }
 
 pub struct RealFileSystem;
@@ -61,4 +69,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

@@ -62,7 +62,7 @@ impl DBEngine {
             return Err(ExecuteError::wrap(error.to_string()));
         }
 
-        // PRIMARY KEY 자동 인덱스 생성 (#217)
+        // PRIMARY KEY 자동 인덱스 생성 (#217, 복합 PK는 #220)
         let primary_key_columns: Vec<String> = if table_info.primary_key.is_empty() {
             table_info
                 .columns
@@ -74,23 +74,22 @@ impl DBEngine {
             table_info.primary_key.clone()
         };
 
-        // TODO(#217): 복합 PRIMARY KEY 인덱스는 미지원 (단일 컬럼만 자동 생성)
-        if primary_key_columns.len() == 1 {
+        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);
             }
 
             let index_name = qualified_index_name(&database_name, &format!("{}_pkey", table_name));
-            let meta = IndexMeta::new(
+            let meta = IndexMeta::new_composite(
                 index_name,
                 table_info.table.clone(),
-                primary_key_columns[0].clone(),
+                primary_key_columns,
                 true,
             );
 
             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/dml/delete.rs

@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
 
 use futures::future::join_all;
 
-use crate::engine::actions::index::row_index_key;
+use crate::engine::actions::index::{join_composite_key, row_index_keys};
 use crate::engine::ast::dml::delete::DeleteQuery;
 use crate::engine::ast::dml::plan::delete::delete_plan::DeletePlanItem;
 use crate::engine::ast::dml::plan::select::scan::ScanType;
@@ -44,7 +44,8 @@ impl DBEngine {
         // WAL-first: 쿼리를 실행/소비하기 전에 페이로드를 미리 직렬화합니다.
         let wal_payload = match &wal_manager {
             Some(_) => Some(
-                bincode::serialize(&query).map_err(|error| ExecuteError::wrap(error.to_string()))?,
+                bincode::serialize(&query)
+                    .map_err(|error| ExecuteError::wrap(error.to_string()))?,
             ),
             None => None,
         };
@@ -140,7 +141,9 @@ impl DBEngine {
             row_indexes.insert(location.row_index);
 
             for meta in &index_metas {
-                if let Some(key) = row_index_key(row, &meta.column_name) {
+                if let Some(key) = row_index_keys(row, &meta.columns)
+                    .map(|components| join_composite_key(&components))
+                {
                     index_removals.push((
                         meta.index_name.clone(),
                         key,

src/engine/actions/dml/insert.rs

@@ -1,6 +1,6 @@
 use std::collections::HashSet;
 
-use crate::engine::actions::index::row_index_key;
+use crate::engine::actions::index::{join_composite_key, row_index_keys};
 use crate::engine::ast::dml::insert::{InsertData, InsertQuery};
 use crate::engine::ast::types::SQLExpression;
 use crate::engine::schema::row::{TableDataField, TableDataRow};
@@ -119,13 +119,12 @@ impl DBEngine {
                         // `query.columns`는 SQL에서 그대로 온 값이라 스키마에 없는
                         // 이름이 들어올 수 있습니다. 파서는 값 개수만 확인하고
                         // 컬럼의 존재 여부는 모릅니다 (#260).
-                        let column_config_info =
-                            columns_map.get(column_name).ok_or_else(|| {
-                                ExecuteError::wrap(format!(
-                                    "column '{}' does not exist on table '{}'",
-                                    column_name, table_name
-                                ))
-                            })?;
+                        let column_config_info = columns_map.get(column_name).ok_or_else(|| {
+                            ExecuteError::wrap(format!(
+                                "column '{}' does not exist on table '{}'",
+                                column_name, table_name
+                            ))
+                        })?;
 
                         let default_value = match &column_config_info.default {
                             Some(default) => default.to_owned(),
@@ -238,7 +237,9 @@ impl DBEngine {
                     let mut batch_keys = HashSet::new();
 
                     for row in &rows {
-                        if let Some(key) = row_index_key(row, &meta.column_name) {
+                        if let Some(key) = row_index_keys(row, &meta.columns)
+                            .map(|components| join_composite_key(&components))
+                        {
                             let duplicated = !self
                                 .index_manager
                                 .get(&meta.index_name, &key)
@@ -249,7 +250,7 @@ impl DBEngine {
                             if duplicated {
                                 return Err(ExecuteError::wrap(format!(
                                     "duplicate key value violates unique index on column '{}'",
-                                    meta.column_name
+                                    meta.column_name()
                                 )));
                             }
                         }
@@ -302,7 +303,9 @@ impl DBEngine {
                     let row_path = (start_index + offset).to_string();
 
                     for meta in &index_metas {
-                        if let Some(key) = row_index_key(row, &meta.column_name) {
+                        if let Some(key) = row_index_keys(row, &meta.columns)
+                            .map(|components| join_composite_key(&components))
+                        {
                             if let Err(error) = self
                                 .index_manager
                                 .insert(&meta.index_name, key.clone(), row_path.clone())

src/engine/actions/dml/update.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 
 use futures::future::join_all;
 
-use crate::engine::actions::index::row_index_key;
+use crate::engine::actions::index::{join_composite_key, row_index_keys};
 use crate::engine::ast::dml::plan::select::scan::ScanType;
 use crate::engine::ast::dml::plan::update::update_plan::UpdatePlanItem;
 use crate::engine::ast::dml::update::UpdateQuery;
@@ -42,7 +42,8 @@ impl DBEngine {
         // WAL-first: 쿼리를 실행/소비하기 전에 페이로드를 미리 직렬화합니다.
         let wal_payload = match &wal_manager {
             Some(_) => Some(
-                bincode::serialize(&query).map_err(|error| ExecuteError::wrap(error.to_string()))?,
+                bincode::serialize(&query)
+                    .map_err(|error| ExecuteError::wrap(error.to_string()))?,
             ),
             None => None,
         };
@@ -179,10 +180,12 @@ impl DBEngine {
                 }
             }
 
-            // 인덱스 컬럼 값 변경 감지 (#217)
+            // 인덱스 컬럼 값 변경 감지 (#217, 복합 키는 #220)
             for meta in &index_metas {
-                let old_key = row_index_key(&old_row, &meta.column_name);
-                let new_key = row_index_key(&row, &meta.column_name);
+                let old_key = row_index_keys(&old_row, &meta.columns)
+                    .map(|components| join_composite_key(&components));
+                let new_key = row_index_keys(&row, &meta.columns)
+                    .map(|components| join_composite_key(&components));
 
                 if old_key != new_key {
                     index_operations.push((
@@ -210,8 +213,7 @@ impl DBEngine {
             }
 
             // 인덱스 반영: 고유 제약 위반은 여기서 검출되며, 실패 시 적용분을 되돌립니다
-            for (i, (index_name, old_key, new_key, row_path)) in
-                index_operations.iter().enumerate()
+            for (i, (index_name, old_key, new_key, row_path)) in index_operations.iter().enumerate()
             {
                 if let Err(error) = self
                     .apply_index_operation(index_name, old_key, new_key, row_path)

src/engine/actions/index.rs

@@ -29,6 +29,43 @@ pub(crate) fn row_index_key(row: &TableDataRow, column_name: &str) -> Option<Str
         })
 }
 
+/// 행의 복수 컬럼 값을 복합 인덱스 키 컴포넌트들로 변환합니다 (#220).
+///
+/// 반환값은 각 컬럼의 `field_to_key` 원본 키 벡터입니다. B-tree에 넣을 단일
+/// 키는 `join_composite_key`로 조합합니다. 조합 시점에 각 컴포넌트가 길이
+/// 프리픽스로 인코딩되어 단사(injective)가 보장되고, 컬럼 순서대로 정렬
+/// 순서가 결정됩니다.
+///
+/// 단일 컬럼 인덱스는 조합 결과가 `field_to_key` 그대로이므로, 기존 단일
+/// 인덱스 키 공간(옵티마이저 eq_key/start_key/end_key와 동일)을 유지합니다.
+///
+/// 인덱스 대상 컬럼 중 하나라도 NULL이거나 행에 없으면 None — 단일 컬럼
+/// 인덱스와 마찬가지로 해당 행은 색인하지 않습니다 (PostgreSQL과 동일).
+pub(crate) fn row_index_keys(row: &TableDataRow, column_names: &[String]) -> Option<Vec<String>> {
+    let mut components = Vec::with_capacity(column_names.len());
+
+    for column_name in column_names {
+        components.push(row_index_key(row, column_name)?);
+    }
+
+    Some(components)
+}
+
+/// 복합 인덱스 키 컴포넌트들을 단일 B-tree 키로 합칩니다 (#220).
+///
+/// 각 컴포넌트를 길이 프리픽스로 인코딩 후 이어붙입니다. 단일 컴포넌트는
+/// 원본 키와 동일한 값이 되어 단일 인덱스의 키 공간과 호환됩니다.
+pub(crate) fn join_composite_key(components: &[String]) -> String {
+    if components.len() == 1 {
+        return components[0].clone();
+    }
+
+    components
+        .iter()
+        .map(|key| super::super::index::encode_composite_key_component(key))
+        .collect()
+}
+
 impl DBEngine {
     /// 서버 기동 후 최초 인덱스 사용 시점에 디스크의 인덱스 파일을 메모리로 적재합니다.
     pub(crate) async fn ensure_indices_loaded(&self) -> errors::Result<()> {
@@ -120,7 +157,8 @@ impl DBEngine {
         let mut distinct_values = HashMap::new();
         for meta in self.table_index_metas(table_name).await {
             if let Ok(distinct) = self.index_manager.distinct_keys(&meta.index_name).await {
-                distinct_values.insert(meta.column_name.clone(), distinct);
+                // 복합 인덱스의 distinct는 첫 컬럼 기준 근사치로 기록 (#220)
+                distinct_values.insert(meta.column_name().to_owned(), distinct);
             }
         }
 

src/engine/actions/mod.rs

@@ -2,3 +2,9 @@ pub mod ddl;
 pub mod dml;
 pub mod etc;
 pub mod index;
+
+#[cfg(test)]
+mod test_composite_index;
+
+#[cfg(test)]
+mod test_composite_pk_e2e;