[기능] 쿼리 히스토리 활용도 강화

[YuujinHwang]

기능 설명 (Description)

• 카탈로그 내 쿼리 히스토리를 활용하여
  • 컬럼의 속성 파악 : Dimension, Fact, Metric, Filter, ...
  • 타 테이블과의 관계 파악 : FK, relationship
  • SQL-to-Lang 을 이용한 QA셋 구축

왜 필요한가요? (Motivation / Use Case)

• 코드/프롬프트는 도메인 지식을 모름 ( 회사마다 다른 경우도 많음 )

  • A라는 컬럼을 집계해야되는데 더해야 할까? 평균을 내야 할까?
  • 설비 가동률을 알려달라는데 24시간으로 나눠야 할까 근무시간으로 나눠야 할까?
  • 재고수량 월마감을 해야하는데 미착품을 인식해야할까 말까?

• 쿼리는 보통 항상 쓰는걸 쓰는 경우가 많음

  • 이번달 매출 / 이익 -> 다음달 매출 / 이익 등
  • 분석 역시 표준화된 Metric들 간의 비교로 출발

제안하는 해결 방법 (Proposed Implementation)

• Table의 하위 document로 Column에 대한 document 강화

  • { "type" : "filter", "statement" : "where A ilike '%text%'", "description" : "품번 조회" }

• 기존 쿼리이력을 긁어 Table간의 끊어진 Relationship 생성

• 기존 쿼리이력과 유사한 경우, 템플릿화된 쿼리를 통해서 즉각 실행

대안 (Alternatives)

• vanna.ai등의 프레임워크 차용
• Q&A RAG셋 구성

추가 고려사항 (Additional Context)

• AI가 생성한 쿼리에 대한 RAG는 어떻게 처리할 것인지 고민이 필요.
• 사용자 영향, 마이그레이션 이슈 등이 있다면

[ehddnr301]

@YuujinHwang 좋은 의견 감사합니다!

제가 이해하기로는 잘 생성한 쿼리를 다음번에도 활용할 수 있는 시스템을 구축하자로 이해했습니다!

LLMOps를 하려면 꼭 필요한 기능이라고 생각됩니다! 혹시 Task를 조금 쪼갤 수 있으면 쪼개서 작업할 수 있으면 좋겠네요!

• Datahub 활용해서 사용하고 등록했던 쿼리를 활용할 수 있도록 기여한 PR이 있긴한데 말씀주신 Feed-Back loop를 구현한 것은 아닙니다아 (현재 활용도 안하고있음)
  Feature/132 metadata 통합 강화 enhancement glossary쿼리예시 메타데이터를 sql 생성 컨텍스트에 반영 #133

[YuujinHwang]

@ehddnr301

현재 Task는 다음과 같이 나뉠 것 같습니다.

1. DDL 및 SQL에서 Named Entity 추출
   1.1 DDL 을 통해 Physical 구조에 대한 Graph생성 ( DB, Schema, Table, Column )
   1.2 SQL을 통해 Logical 레이어 Graph 생성 ( View, Metric )
   1.3 SQL을 통해 Semantic 레이어 Graph 생성 -> Query 이력의 패턴화

2. Graph Node및 edge에 description 생성 및 embedding

3. Retrieval 전략 생성

최종적인 목표는 사람들이 이미 사용하고 있는 '의미있는' 쿼리들을 지속적으로 업데이트하고
LLM이 유사한 목적의 쿼리를 보다 더 잘 만들수 있게 하는 것입니다.

테스트해보고 있는 구조는 아래와 같습니다.

class Contains(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    parent_id: str
    parent_type: Literal['database', 'schema', 'table']
    child_id: str
    child_type: Literal['schema', 'table', 'column', 'metric']
    created_at: datetime = Field(default_factory=datetime.now)

    @field_validator('child_type')
    @classmethod
    def validate_hierarchy(cls, v, info):
        hierarchy = {
            'database': {'schema'},
            'schema': {'table'},
            'table': {'column', 'metric'}
        }
        parent_type = info.data.get('parent_type')
        if parent_type and v not in hierarchy.get(parent_type, set()):
            raise ValueError(f"Invalid child_type '{v}' for parent_type '{parent_type}'")
        return v
    
# Join -> Table
class JoinsWith(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    join_id: str
    table_id: str
    direction: Literal['left', 'right', 'undirected']
    created_at: datetime = Field(default_factory=datetime.now)

# Join -> Column
class JoinsOn(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    join_id: str
    column_id: str
    direction: Literal['left', 'right', 'undirected']
    created_at: datetime = Field(default_factory=datetime.now)

# Union -> Table
class UnionsWith(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    union_id: str
    table_id: str
    created_at: datetime = Field(default_factory=datetime.now)

# QueryPattern -> Table
class AccessesTo(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    source_id: str
    source_label: Literal['QueryPattern', 'Table'] = 'QueryPattern'
    table_id: str
    created_at: datetime = Field(default_factory=datetime.now)

# QueryPattern -> Column,
class UsesColumn(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    query_pattern_id: str
    column_id: str
    usage: List[Literal['where', 'group_by', 'select', 'order_by', 'join', 'other']]
    usage_count: List[int] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.now)

    @field_validator('usage', mode='after')
    @classmethod
    def ensure_unique_usage(cls, values: List[str]) -> List[str]:
        seen = []
        for value in values:
            if value not in seen:
                seen.append(value)
        return seen

    @field_validator('usage_count', mode='after')
    @classmethod
    def ensure_count_alignment(cls, counts: List[int], info) -> List[int]:
        usage_values = info.data.get('usage', [])
        if not counts:
            return [1 for _ in usage_values]
        if len(counts) != len(usage_values):
            raise ValueError("usage_count length must match usage length")
        return [int(count) for count in counts]

# QueryPattern -> Metric
class UsesMetric(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    query_pattern_id: str
    metric_id: str
    operation: Optional[Literal['sum', 'avg', 'min', 'max', 'count']] = None
    created_at: datetime = Field(default_factory=datetime.now)

# Metric -> Column, Metric -> Metric (Derived Metric)
class BasedOn(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    metric_id: str
    based_on_id: str  # can be column_id or metric_id
    operation: Optional[Literal['sum', 'avg', 'min', 'max', 'count']] = None
    created_at: datetime = Field(default_factory=datetime.now)

# Query -> QueryPattern
class FollowsPattern(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    query_id: str
    query_pattern_id: str
    created_at: datetime = Field(default_factory=datetime.now)

# Graph Nodes
class Database(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))

class Schema(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))

class Table(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    layer: Optional[Literal['bronze', 'silver', 'gold']] = None
    description: str
    description_embedding: List[float]
    type: Literal['fact', 'dimension']
    physical_type: Literal['table', 'view', 'materialized_view']
    ddl: Optional[str] = None
    ddl_embedding: Optional[List[float]] = None
    source_tables: List[str] = Field(default_factory=list)
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

class Column(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    description: str
    description_embedding: List[float]
    type: Literal[
        "bigint",
        "binary",
        "boolean",
        "date",
        "decimal",
        "double",
        "float",
        "int",
        "interval",
        "void",
        "smallint",
        "string",
        "timestamp",
        "timestamp_ntz",
        "tinyint",
        "geography",
        "geometry",
        "array",
        "map",
        "struct",
        "variant",
        "object"
    ]
    sample_values: Optional[List[str]]
    parser_sql: Optional[str] = None
    parser_description: Optional[str] = None
    
class Join(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    type: Literal['inner', 'left', 'right', 'full', 'cross', 'self']
    condition_sql: str
    description: str = None
    description_embedding: List[float] = None
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

class Union(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    type: Literal['union', 'union_all']
    description: str = None
    description_embedding: List[float] = None
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

class Query(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    intent: Optional[str]
    intent_embedding: Optional[List[float]]
    sql: str
    description: str = None
    description_embedding: List[float] = None
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

class QueryPattern(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    intent: Optional[str]
    intent_embedding: Optional[List[float]]
    parameterized_sql: str
    parameterized_sql_embedding: List[float]
    parameters: List[dict[str, str]]
    description: str = None
    description_embedding: List[float] = None
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

class Metric(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    description: str
    description_embedding: List[float]
    aggregation: Literal['sum', 'avg', 'min', 'max', 'count', 'expression']
    expression_sql: str
    expression_sql_embedding: List[float]
    access_frequency: int
    created_at: datetime = Field(default_factory=datetime.now)

[ehddnr301]

@YuujinHwang 너무 좋고 발전가능성이 많은 아이디어같습니다! 관련해서 월요일 스터디시간에 다른분들도 같이 의견나눠보면 좋을것 같아요!

@CausalInferenceLab/lang2sql 의견 많이주세요!

[YuujinHwang]

코드변경사항이 매우 커지고있어 이슈는 History Fetch -> Normalized Query 생성 까지로 짤라서 커밋 PR하고
Neo4j 구성 / datahub 커넥션 / DB구조 추출 / 리트리벌 로 쪼개서 후속이슈로 처리하면 좋을 것 같은데 의견 부탁드립니다. @ehddnr301

[ehddnr301]

@YuujinHwang 너무 좋습니다! 진행하시는것 공유주시거나 PR생성해주시면 매우 열심히 공부하겠습니다!