#lang racket

(provide (struct-out column-info)
         (struct-out table)
         (struct-out and-f)
         (struct-out or-f)
         (struct-out not-f)
         (struct-out eq-f)
         (struct-out eq2-f)
         (struct-out lt-f)
         table-insert
         table-project
         table-sort
         table-select
         table-rename
         table-cross-join
         table-natural-join)

(define-struct column-info (name type) #:transparent)

(define-struct table (schema rows) #:transparent)

(define cities
  (table
   (list (column-info 'city    'string)
         (column-info 'country 'string)
         (column-info 'area    'number)
         (column-info 'capital 'boolean))
   (list (list "Wrocław" "Poland"  293 #f)
         (list "Warsaw"  "Poland"  517 #t)
         (list "Poznań"  "Poland"  262 #f)
         (list "Berlin"  "Germany" 892 #t)
         (list "Munich"  "Germany" 310 #f)
         (list "Paris"   "France"  105 #t)
         (list "Rennes"  "France"   50 #f))))

(define countries
  (table
   (list (column-info 'country 'string)
         (column-info 'population 'number))
   (list (list "Poland" 38)
         (list "Germany" 83)
         (list "France" 67)
         (list "Spain" 47))))

(define (empty-table columns) (table columns '()))

; Wstawianie

(define (val-type-ok? val type)
  (cond ([and (eq? type 'number) (number? val)] true)
        ([and (eq? type 'string) (string? val)] true)
        ([and (eq? type 'symbol) (symbol? val)] true)
        ([and (eq? type 'boolean) (boolean? val)] true)
        (else false)))

(define (type-ok? row schema)
  (cond ([and (empty? row) (empty? schema)] true)
        ([or (empty? row) (empty? schema)] false)
        ([val-type-ok? (car row) (column-info-type (car schema))] (type-ok? (cdr row) (cdr schema)))
        (else false)))

(define (table-insert row tab)
  (if (type-ok? row (table-schema tab))
      (table (table-schema tab) (cons row (table-rows tab)))
      (error "invalid row types")))


; Projekcja
(define (column-name-to-id name schema)
  (define (cmp-names column name) (equal? name (column-info-name column)))
  (index-of schema name cmp-names))

(define (column-names-to-ids names schema)
  (map (lambda (n) (column-name-to-id n schema)) names))

(define (table-project xs tab)
  (define ids (column-names-to-ids xs (table-schema tab)))
  (define schema (map (lambda (id) (list-ref (table-schema tab) id)) ids))
  (define rows (map (lambda (row)
    (map (lambda (id) (list-ref row id)) ids))
  (table-rows tab)))
  (table schema rows))


;; Sortowanie
(define (compare val1 val2)
  (cond ([integer? val1] (< val1 val2))
        ([string? val1] (string<? val1 val2))
        ([symbol? val1] (symbol<? val1 val2))
        (else (and (not val1) val2))))

(define (sort1 col tab)
  (define id (column-name-to-id col (table-schema tab)))
  (define rows (sort (table-rows tab) (lambda (x y) (compare (list-ref x id) (list-ref y id)))))
  (table (table-schema tab) rows))

(define (table-sort cols tab)
  (if (empty? cols)
      tab
      (sort1 (car cols) (table-sort (cdr cols) tab))))
;; Selekcja

(define-struct and-f (l r))
(define-struct or-f (l r))
(define-struct not-f (e))
(define-struct eq-f (name val))
(define-struct eq2-f (name name2))
(define-struct lt-f (name val))
;
(define (matches? formula schema row)
  (match formula
    ([and-f l r] (and (matches? l schema row) (matches? r schema row)))
    ([or-f l r] (or (matches? l schema row) (matches? r schema row)))
    ([not-f e] (not (matches? e schema row)))
    ([eq-f name val] (equal? (list-ref row (column-name-to-id name schema)) val))
    ([eq2-f name name2] (equal? (list-ref row (column-name-to-id name schema)) (list-ref row (column-name-to-id name2 schema))))
    ([lt-f name val] (compare (list-ref row (column-name-to-id name schema)) val))
    (else false)))

(define (table-select formula tab)
  (define schema (table-schema tab))
  (define rows (filter (lambda (row) (matches? formula schema row)) (table-rows tab)))
  (table schema rows))


;; Zmiana nazwy
(define (table-rename old-name new-name tab)
  (define (rename-column column)
    (if (eq? (column-info-name column) old-name)
        (column-info new-name (column-info-type column))
        column))
  (define schema (map rename-column (table-schema tab)))
  (table schema (table-rows tab)))


;; Złączenie kartezjańskie

(define (table-cross-join tab1 tab2)
  (define schema (append (table-schema tab1) (table-schema tab2)))
  (define rows (append-map
    (lambda (row1) (map
      (lambda (row2) (append row1 row2))
      (table-rows tab2)))
    (table-rows tab1)))
  (table schema rows))

; Złączenie

(define (table-natural-join tab1 tab2)
 (define columns1 (map column-info-name (table-schema tab1)))
  (define columns2 (map column-info-name (table-schema tab2)))
  (define column-intersection (set-intersect columns1 columns2))
  (define (add2 sym) (string->symbol (string-append (symbol->string sym) "2")))
  (define column-intersection2 (map add2 column-intersection))
  (define (add2-tab xs tab)
    (if (empty? xs)
        tab
        (add2-tab (cdr xs) (table-rename (car xs) (add2 (car xs)) tab))))
  (define tab3 (table-cross-join tab1 (add2-tab column-intersection tab2)))
  (define columns3 (map column-info-name (table-schema tab3)))
  (define ids1 (column-names-to-ids column-intersection (table-schema tab3)))
  (define ids2 (column-names-to-ids column-intersection2 (table-schema tab3)))
  (define (row-ok? row) (andmap (lambda (id1 id2) (equal? (list-ref row id1) (list-ref row id2))) ids1 ids2))
  (define tab4 (table (table-schema tab3) (filter row-ok? (table-rows tab3))))
  (table-project (set-subtract columns3 column-intersection2) tab4))