/*
 * Copyright 2013 Yuichiro Moriguchi
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.morilib.awk.builtin.rel;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import net.morilib.awk.expr.AwkExpression;
import net.morilib.awk.io.AwkFiles;
import net.morilib.awk.namespace.AwkFunctionNamespace;
import net.morilib.awk.namespace.AwkNamespace;
import net.morilib.awk.value.AwkArray;
import net.morilib.awk.value.AwkString;
import net.morilib.awk.value.AwkUndefined;
import net.morilib.awk.value.AwkValue;

/**
 * リレーショナル演算をするユーティリティメソッドです。
 *
 *
 * @author MORIGUCHI, Yuichiro 2013/04/13
 */
public final class Relations {

	//
	static final String TBLSEP = "TBLSEP";

	//
	static final String SUBSEP = "SUBSEP";

	private static final Pattern P1 =
		Pattern.compile("^(.*)([0-9]+)#");

	private Relations() {}

	/**
	 * 射影を計算します。
	 * 
	 * @param table     テーブル
	 * @param namespace 名前空間
	 * @param records   射影対象のレコード
	 * @return 射影されたテーブル
	 */
	public static AwkArray projection(AwkArray table,
			AwkNamespace namespace,
			String... records) {
		StringBuffer b;
		AwkArray r = new AwkArray();
		String d, sep, nsp;
		String[] rd;

		if(records == null)  throw new NullPointerException();
		if(records.length == 0) {
			throw new IllegalArgumentException();
		}

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> z : table.entrySet()) {
			b  = new StringBuffer();
			d  = "";
			rd = z.getValue().toString(namespace).split(sep);
			for(String s : rd) {
				for(String t : records) {
					if(s.startsWith(t + nsp)) {
						b.append(d).append(s);
						d = sep;
					}
				}
			}
			r.putArray(z.getKey(), AwkString.valueOf(b.toString()));
		}
		return r;
	}

	private static String getnm(String s) {
		Matcher m = P1.matcher(s);

		if(m.matches()) {
			return m.group(1) + (Integer.parseInt(m.group(2)) + 1);
		} else {
			return s + "1";
		}
	}

	/**
	 * 直積を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 直積
	 */
	public static AwkArray crossJoin(AwkArray table1, AwkArray table2,
			AwkNamespace namespace) {
		AwkArray r = new AwkArray();
		String[] p1, q1, t0;
		String x, sep, nsp;
		StringBuffer b;
		int c = 1;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> v : table1.entrySet()) {
			p1 = (x = v.getValue().toString(namespace)).split(sep);
			for(Map.Entry<String, AwkValue> w : table2.entrySet()) {
				q1 = w.getValue().toString(namespace).split(sep);
				b  = new StringBuffer(x);
				l1: for(int i = 0; i < q1.length; i++) {
					t0 = q1[i].split(nsp);
					for(int j = 0; j < p1.length; j++) {
						if(p1[j].startsWith(t0[0] + nsp)) {
							b.append(sep);
							b.append(getnm(t0[0]));
							b.append(nsp);
							b.append(t0[1]);
							continue l1;
						}
					}
					b.append(sep).append(q1[i]);
				}
				r.putArray(c++ + "", AwkString.valueOf(b.toString()));
			}
		}
		return r;
	}

	private static String _find(String x, String[] p1, String[] q1,
			String sep, String nsp, AwkNamespace ns) {
		Map<String, String> m = new HashMap<String, String>();
		StringBuffer b = new StringBuffer(x);
		String[] tp, tq;
		String y;

		for(int j = 0; j < p1.length; j++) {
			tp = p1[j].split(nsp);
			m.put(tp[0], tp[1]);
		}

		for(int i = 0; i < q1.length; i++) {
			tq = q1[i].split(nsp);
			if((y = m.get(tq[0])) == null) {
				b.append(sep).append(q1[i]);
			} else if(y.equals(tq[1])) {
				// do nothing
			} else {
				return null;
			}
		}
		return b.toString();
	}

	/**
	 * 自然結合を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 自然結合
	 */
	public static AwkArray naturalJoin(AwkArray table1,
			AwkArray table2, AwkNamespace namespace) {
		AwkArray r = new AwkArray();
		String x, y, sep, nsp;
		String[] p1, q1;
		int c = 1;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> v : table1.entrySet()) {
			p1 = (x = v.getValue().toString(namespace)).split(sep);
			for(Map.Entry<String, AwkValue> w : table2.entrySet()) {
				q1 = w.getValue().toString(namespace).split(sep);
				y  = _find(x, p1, q1, sep, nsp, namespace);
				if(y != null) {
					r.putArray(c++ + "", AwkString.valueOf(y));
				}
			}
		}
		return r;
	}

	private static Set<String> _find2(String[] p1, String[] q1,
			String sep, String nsp, AwkNamespace ns,
			Map<String, Set<String>> uni) {
		Map<String, String> m = new HashMap<String, String>();
		Set<String> ss, sq;
		boolean r = false;
		String[] tp, tq;
		String y;

		for(int j = 0; j < p1.length; j++) {
			tp = p1[j].split(nsp);
			m.put(tp[0], tp.length < 2 ? "" : tp[1]);
		}

		sq = new HashSet<String>();
		for(int i = 0; i < q1.length; i++) {
			tq = q1[i].split(nsp);
			sq.add(tq[0]);
			if((y = m.get(tq[0])) == null) {
				// do nothing
			} else if(!y.equals(tq[1])) {
				return null;
			} else if((ss = uni.get(tq[0])) == null ||
					!ss.contains(y)) {
				if(ss == null) {
					ss = new HashSet<String>();
					uni.put(tq[0], ss);
				}
				ss.add(y);
				r = true;
			} else {
				r = r || false;
			}
		}
		return r && m.keySet().containsAll(sq) ? sq : null;
	}

	/**
	 * 制限を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 制限
	 */
	public static AwkArray restriction(AwkArray table1,
			AwkArray table2, AwkNamespace namespace) {
		AwkArray r = new AwkArray();
		Map<String, Set<String>> m;
		String[] p1, q1;
		String sep, nsp;
		AwkValue x;
		int c = 1;

		m   = new HashMap<String, Set<String>>();
		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> v : table1.entrySet()) {
			p1 = (x = v.getValue()).toString(namespace).split(sep);
			for(Map.Entry<String, AwkValue> w : table2.entrySet()) {
				q1 = w.getValue().toString(namespace).split(sep);
				if(_find2(p1, q1, sep, nsp, namespace, m) != null) {
					r.putArray(c++ + "", x);
				}
			}
		}
		return r;
	}

	/**
	 * 選択を計算します。
	 * 
	 * @param table テーブル
	 * @param f     式の中間表現
	 * @param namespace 名前空間
	 * @param files     ファイル管理
	 * @return 選択
	 */
	public static AwkArray select(AwkArray table, AwkExpression f,
			AwkNamespace namespace, AwkFiles files) {
		AwkArray r = new AwkArray();
		AwkNamespace ns;
		String[] p1, tp;
		String sep, nsp;
		int c = 1;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(AwkValue v : table.values()) {
			ns = new AwkFunctionNamespace(namespace);
			p1 = v.toString(namespace).split(sep);
			for(int j = 0; j < p1.length; j++) {
				tp = p1[j].split(nsp);
				ns.assign(tp[0], AwkValue.cast(
						tp.length < 2 ? "" : tp[1]));
			}

			if(f.eval(ns, files).toBoolean(namespace)) {
				r.putArray(c++ + "", v);
			}
		}
		return r;
	}

	private static Set<Map<String, String>> toSetOfMap(
			AwkArray table, AwkNamespace ns, String sep, String nsp) {
		Set<Map<String, String>> t;
		Map<String, String> m;
		String[] p, q;

		t = new HashSet<Map<String, String>>();
		for(AwkValue v : table.values()) {
			m = new HashMap<String, String>();
			p = v.toString(ns).split(sep);
			for(String s : p) {
				q = s.split(nsp);
				m.put(q[0], q.length < 2 ? "" : q[1]);
			}
			t.add(m);
		}
		return t;
	}

	private static AwkArray toArrayOfString(Set<Map<String, String>> t,
			String sep, String nsp) {
		AwkArray a = new AwkArray();
		StringBuffer b;
		int c = 1;
		String d;

		for(Map<String, String> m : t) {
			b = new StringBuffer();
			d = "";
			for(Map.Entry<String, String> z : m.entrySet()) {
				b.append(d).append(z.getKey()).append(nsp);
				b.append(z.getValue());
				d = sep;
			}
			a.putArray(c++ + "", AwkString.valueOf(b.toString()));
		}
		return a;
	}

	/**
	 * 和集合を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 和集合
	 */
	public static AwkArray union(AwkArray table1, AwkArray table2,
			AwkNamespace namespace) {
		Set<Map<String, String>> t1, t2;
		String sep, nsp;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		t1 = toSetOfMap(table1, namespace, sep, nsp);
		t2 = toSetOfMap(table2, namespace, sep, nsp);
		t1.addAll(t2);
		return toArrayOfString(t1, sep, nsp);
	}

	/**
	 * 共通部分を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 共通部分
	 */
	public static AwkArray intersect(AwkArray table1, AwkArray table2,
			AwkNamespace namespace) {
		Set<Map<String, String>> t1, t2;
		String sep, nsp;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		t1 = toSetOfMap(table1, namespace, sep, nsp);
		t2 = toSetOfMap(table2, namespace, sep, nsp);
		t1.retainAll(t2);
		return toArrayOfString(t1, sep, nsp);
	}

	/**
	 * 差集合を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 差集合
	 */
	public static AwkArray except(AwkArray table1, AwkArray table2,
			AwkNamespace namespace) {
		Set<Map<String, String>> t1, t2;
		String sep, nsp;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		t1 = toSetOfMap(table1, namespace, sep, nsp);
		t2 = toSetOfMap(table2, namespace, sep, nsp);
		t1.removeAll(t2);
		return toArrayOfString(t1, sep, nsp);
	}

	/**
	 * 商集合を計算します。
	 * 
	 * @param table1 テーブル1
	 * @param table2 テーブル2
	 * @param namespace 名前空間
	 * @return 商集合
	 */
	public static AwkArray divide(AwkArray table1,
			AwkArray table2, AwkNamespace namespace) {
		Set<Map<String, String>> a = null, b;
		Map<String, Set<String>> m;
		Map<String, String> o;
		String[] p1, q1, t1;
		String sep, nsp;
		Set<String> rq;

		m   = new HashMap<String, Set<String>>();
		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> v : table2.entrySet()) {
			p1 = v.getValue().toString(namespace).split(sep);
			b  = new HashSet<Map<String, String>>();
			for(Map.Entry<String, AwkValue> w : table1.entrySet()) {
				m.clear();
				o  = new HashMap<String, String>();
				q1 = w.getValue().toString(namespace).split(sep);
				rq = _find2(q1, p1, sep, nsp, namespace, m);
				if(rq != null) {
					for(int i = 0; i < q1.length; i++) {
						t1 = q1[i].split(nsp);
						if(!rq.contains(t1[0])) {
							o.put(t1[0], t1[1]);
						}
					}
					b.add(o);
				}
			}

			if(a == null) {
				a = b;
			} else {
				a.retainAll(b);
			}
		}
		return toArrayOfString(a, sep, nsp);
	}

	/**
	 * タプルを生成します。
	 * 
	 * @param a 配列
	 * @param namespace 名前空間
	 * @return タプル
	 */
	public static AwkValue makeTuple(AwkArray a,
			AwkNamespace namespace) {
		StringBuffer b = new StringBuffer();
		String sep, nsp, d = "";

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		for(Map.Entry<String, AwkValue> z : a.entrySet()) {
			b.append(d).append(z.getKey()).append(nsp);
			b.append(z.getValue().toString(namespace));
			d = sep;
		}
		return AwkString.valueOf(b.toString());
	}

	/**
	 * タプルから配列を生成します。
	 * 
	 * @param a タプル
	 * @param namespace 名前空間
	 * @return 配列
	 */
	public static AwkArray makeArray(AwkValue x,
			AwkNamespace namespace) {
		AwkArray a = new AwkArray();
		String sep, nsp;
		String[] p, q;

		sep = namespace.getRoot().find(SUBSEP).toString(namespace);
		nsp = namespace.getRoot().find(TBLSEP).toString(namespace);
		p = x.toString(namespace).split(sep);
		for(String s : p) {
			q = s.split(nsp);
			if(q.length < 2) {
				a.putArray(q[0], AwkUndefined.UNDEF);
			} else {
				a.putArray(q[0], AwkValue.cast(q[1]));
			}
		}
		return a;
	}

}
