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

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,18 +74,17 @@ 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;
                 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,
             );
 

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;

src/engine/actions/test_composite_index.rs

@@ -0,0 +1,149 @@
+//! Task 2 RED 테스트: 복합 인덱스 키 지원 (#220)
+//!
+//! - IndexMeta.columns (복수 컬럼) 지원
+//! - row_index_keys: 복수 컬럼 키 인코딩 (순서 보존, 단사)
+//! - NULL 포함 컬럼 → 인덱스 제외 (PostgreSQL과 동일)
+
+use crate::engine::actions::index::row_index_keys;
+use crate::engine::ast::types::TableName;
+use crate::engine::index::{IndexMeta, field_to_key};
+use crate::engine::schema::row::{TableDataField, TableDataFieldType, TableDataRow};
+
+fn table() -> TableName {
+    TableName::new(Some("rrdb".to_owned()), "memberships".to_owned())
+}
+
+fn field(name: &str, data: TableDataFieldType) -> TableDataField {
+    TableDataField {
+        table_name: table(),
+        column_name: name.to_owned(),
+        data,
+    }
+}
+
+#[test]
+fn index_meta_supports_multiple_columns() {
+    let meta = IndexMeta::new_composite(
+        "rrdb.memberships_pkey".to_owned(),
+        table(),
+        vec!["user_id".to_owned(), "group_id".to_owned()],
+        true,
+    );
+
+    assert_eq!(meta.columns(), vec!["user_id", "group_id"]);
+    // 하위 호환: 단일 컬럼 인덱스는 column_name()로 첫 컬럼을 반환
+    let single = IndexMeta::new("rrdb.users_pkey".to_owned(), table(), "id".to_owned(), true);
+    assert_eq!(single.columns(), vec!["id"]);
+    assert_eq!(single.column_name(), "id");
+}
+
+#[test]
+fn composite_key_is_concatenation_with_unambiguous_boundaries() {
+    let row = TableDataRow {
+        fields: vec![
+            field("user_id", TableDataFieldType::Integer(1)),
+            field("group_id", TableDataFieldType::Integer(2)),
+        ],
+    };
+
+    let keys = row_index_keys(&row, &["user_id".to_owned(), "group_id".to_owned()]).unwrap();
+
+    // 두 키를 합친 값이 단사여야 함: 다른 컬럼 조합이 같은 키를 만들 수 없음
+    let combined = keys.join("");
+    assert!(combined.contains(&field_to_key(&TableDataFieldType::Integer(1))));
+    assert!(combined.contains(&field_to_key(&TableDataFieldType::Integer(2))));
+}
+
+#[test]
+fn composite_key_encoding_is_injective_across_value_shapes() {
+    use crate::engine::actions::index::join_composite_key;
+
+    // "S:ab" + "S:c" vs "S:a" + "S:bc" 가 같은 키가 되는 충돌을 막아야 함
+    let row_a = TableDataRow {
+        fields: vec![
+            field("x", TableDataFieldType::String("ab".to_owned())),
+            field("y", TableDataFieldType::String("c".to_owned())),
+        ],
+    };
+    let row_b = TableDataRow {
+        fields: vec![
+            field("x", TableDataFieldType::String("a".to_owned())),
+            field("y", TableDataFieldType::String("bc".to_owned())),
+        ],
+    };
+
+    let keys_a = row_index_keys(&row_a, &["x".to_owned(), "y".to_owned()]).unwrap();
+    let keys_b = row_index_keys(&row_b, &["x".to_owned(), "y".to_owned()]).unwrap();
+
+    assert_ne!(
+        keys_a, keys_b,
+        "different value splits must not collide: {:?} vs {:?}",
+        keys_a, keys_b
+    );
+
+    // 조합된 B-tree 키도 충돌 없어야 함 (길이 프리픽스 인코딩)
+    assert_ne!(
+        join_composite_key(&keys_a),
+        join_composite_key(&keys_b),
+        "joined composite keys must not collide"
+    );
+}
+
+#[test]
+fn composite_key_preserves_column_order() {
+    let row = TableDataRow {
+        fields: vec![
+            field("a", TableDataFieldType::Integer(1)),
+            field("b", TableDataFieldType::Integer(2)),
+        ],
+    };
+
+    let ab = row_index_keys(&row, &["a".to_owned(), "b".to_owned()]).unwrap();
+    let ba = row_index_keys(&row, &["b".to_owned(), "a".to_owned()]).unwrap();
+
+    assert_ne!(ab, ba, "column order must affect the key");
+    // 컴포넌트는 field_to_key 원본 키
+    assert_eq!(ab[0], field_to_key(&TableDataFieldType::Integer(1)));
+    assert_eq!(ab[1], field_to_key(&TableDataFieldType::Integer(2)));
+    assert_eq!(ba[0], field_to_key(&TableDataFieldType::Integer(2)));
+}
+
+#[test]
+fn row_with_null_in_indexed_column_is_not_indexed() {
+    let row = TableDataRow {
+        fields: vec![
+            field("user_id", TableDataFieldType::Integer(1)),
+            field("group_id", TableDataFieldType::Null),
+        ],
+    };
+
+    assert!(
+        row_index_keys(&row, &["user_id".to_owned(), "group_id".to_owned()]).is_none(),
+        "NULL이 포함된 복합 키는 색인하지 않음 (PostgreSQL과 동일)"
+    );
+}
+
+#[test]
+fn row_missing_indexed_column_is_not_indexed() {
+    let row = TableDataRow {
+        fields: vec![field("user_id", TableDataFieldType::Integer(1))],
+    };
+
+    assert!(row_index_keys(&row, &["user_id".to_owned(), "missing".to_owned()]).is_none());
+}
+
+#[test]
+fn test_single_column_row_index_keys_matches_row_index_key() {
+    use crate::engine::actions::index::{join_composite_key, row_index_key, row_index_keys};
+
+    let row = TableDataRow {
+        fields: vec![field("id", TableDataFieldType::Integer(42))],
+    };
+
+    let via_multi = row_index_keys(&row, &["id".to_owned()]).unwrap();
+    let via_single = row_index_key(&row, "id").unwrap();
+
+    // 단일 컬럼 복합 키를 join하면 기존 row_index_key와 정확히 일치해야 함:
+    // 단일 인덱스 경로가 동일 키 공간(옵티마이저 eq_key 포함)을 유지하는 계약 (#220)
+    assert_eq!(join_composite_key(&via_multi), via_single);
+}