@Override
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

		// 讀取一列轉換為欄位陣列
		String[] columnValue = readRow(value, VERTICALBAR);

		// 建立Key <primaryKey, tableIndex, joinCondition>
		outputKey = new DoubleTextPair(new IntWritable(Integer.parseInt(columnValue[0])), new IntWritable(tableIndex));
		boolean result = false;

		// find column & check it
		for (String c : filter) {
			for (int i = 0; i < SCHEMA[tableIndex].length; i++) {
				if (c.contains(SCHEMA[tableIndex][i])) {
					result = conditionValidator.valid(tableIndex, i, columnValue[i]);
				}
			}
		}
		if (result) {
			// 建立Value <PK, PK, PK, ..., PK>
			StringBuffer sb = new StringBuffer();
			for (String c : column) {// match column
				for (int i = 0; i < SCHEMA[tableIndex].length; i++) {
					if (c.equals(SCHEMA[tableIndex][i])) {
						sb.append(columnValue[i]);
						sb.append(COMMA + WHITE_SPACE);
					}
				}
			}
			if (sb.length() > 2) {
				outputValue.set(sb.substring(0, sb.length() - 2));
			}
			context.write(outputKey, outputValue);
		}
	}