Project

General

Profile

1
package eu.dnetlib.data.transform;
2

    
3
import java.io.*;
4
import java.util.*;
5
import java.util.Map.Entry;
6
import java.util.zip.GZIPInputStream;
7
import javax.xml.transform.TransformerConfigurationException;
8
import javax.xml.transform.TransformerFactoryConfigurationError;
9

    
10
import com.google.common.base.Function;
11
import com.google.common.collect.Iterables;
12
import com.google.common.collect.Lists;
13
import com.google.common.collect.Maps;
14
import com.google.common.collect.Sets;
15
import com.google.protobuf.InvalidProtocolBufferException;
16
import com.googlecode.protobuf.format.JsonFormat;
17
import com.googlecode.protobuf.format.JsonFormat.ParseException;
18
import eu.dnetlib.actionmanager.actions.ActionFactory;
19
import eu.dnetlib.actionmanager.actions.XsltInfoPackageAction;
20
import eu.dnetlib.actionmanager.common.Agent;
21
import eu.dnetlib.actionmanager.common.Operation;
22
import eu.dnetlib.actionmanager.common.Provenance;
23
import eu.dnetlib.data.mapreduce.hbase.index.config.*;
24
import eu.dnetlib.data.mapreduce.util.*;
25
import eu.dnetlib.data.proto.KindProtos.Kind;
26
import eu.dnetlib.data.proto.OafProtos.Oaf;
27
import eu.dnetlib.data.proto.TypeProtos.Type;
28
import eu.dnetlib.miscutils.functional.xml.IndentXmlString;
29
import org.apache.commons.io.IOUtils;
30
import org.apache.commons.lang.StringUtils;
31
import org.apache.commons.logging.Log;
32
import org.apache.commons.logging.LogFactory;
33
import org.dom4j.Document;
34
import org.dom4j.DocumentException;
35
import org.dom4j.io.SAXReader;
36
import org.json.JSONObject;
37
import org.junit.Before;
38
import org.junit.Ignore;
39
import org.junit.Test;
40
import org.springframework.core.io.ByteArrayResource;
41
import org.springframework.core.io.Resource;
42

    
43
import static org.junit.Assert.*;
44

    
45
public class XsltRowTransformerFactoryTest {
46

    
47
	private static final Log log = LogFactory.getLog(XsltRowTransformerFactoryTest.class);
48
	private static String basePathProfiles = "/eu/dnetlib/test/profiles/TransformationRuleDSResources/TransformationRuleDSResourceType/2hbase/";
49
	private XsltRowTransformerFactory factory;
50
	private EntityConfigTable entityConfigTable;
51

    
52
	@Before
53
	public void setUp() throws Exception {
54
		factory = new XsltRowTransformerFactory();
55
		entityConfigTable = IndexConfig.load(IndexConfigTest.config).getConfigMap();
56
	}
57

    
58
	@Test
59
	@Ignore // need to reimplement because claimUpdates_2_hbase.xsl was removed
60
	public void testParseOafClaimUpdate() throws Exception {
61
		doTest(loadFromTransformationProfile("claimUpdates_2_hbase.xsl"), load("recordClaimUpdate.xml"));
62
	}
63

    
64
	@Test
65
	@Ignore // need to reimplement because claimUpdates_2_hbase.xsl was removed
66
	public void testParseClaimUpdate() throws Exception {
67

    
68
		final List<Row> rows = Lists.newArrayList();
69
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordClaim.xml")));
70
		rows.addAll(asRows(loadFromTransformationProfile("claimUpdates_2_hbase.xsl"), load("recordClaimUpdate.xml")));
71

    
72
		printAll(mapAll(buildTable(rows)));
73
	}
74

    
75
	@Test
76
	public void testParseClaimRel() throws Exception {
77

    
78
		doTest(loadFromTransformationProfile("claimRels_2_hbase.xml"), load("recordClaimRel.xml"));
79
	}
80

    
81

    
82
	@Test
83
	public void testParseFp7IctPUB() throws Exception {
84

    
85
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("ec_fp7_ict.xml"));
86
	}
87

    
88
	@Test
89
	public void testParseRecordCrossref() throws Exception {
90

    
91
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordCrossref.xml"));
92
	}
93

    
94
	@Test
95
	public void testParseDatasetPUB() throws Exception {
96

    
97
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatasetPUB.xml"));
98
	}
99

    
100
	@Test
101
	public void testParseSoftwareEgiApp() throws Exception {
102

    
103
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("softwareEgiApp.xml"));
104
	}
105

    
106
	@Test
107
	public void testParseSoftwareEgiApp2() throws Exception {
108

    
109
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("softwareEgiApp2.xml"));
110
	}
111

    
112
	@Test
113
	public void testParseOrpEgiApp() throws Exception {
114

    
115
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("orpEgiApp.xml"));
116
	}
117

    
118
	@Test
119
	public void testParseSoftwareDOECODE() throws Exception {
120

    
121
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("doecode.xml"));
122
	}
123

    
124

    
125
	@Test
126
	public void testParseDatasetLindat() throws Exception {
127

    
128
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("datasetLindat.xml"));
129
	}
130

    
131
	@Test
132
	public void testParseDatasetNeuroVault() throws Exception {
133

    
134
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordNeuroVault.xml"));
135
	}
136

    
137
	@Test
138
	public void testParseDatasetNeuroVault2() throws Exception {
139

    
140
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordNeuroVault2.xml"));
141
	}
142

    
143
	@Test
144
	public void testParseClaim() throws Exception {
145

    
146
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordClaim.xml"));
147
	}
148

    
149
	@Test
150
	public void testParseClaimDedup() throws Exception {
151

    
152
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordClaimedDedup.xml"));
153
	}
154

    
155

    
156
	@Test
157
	public void testParseClaimDataset() throws Exception {
158

    
159
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordClaimDataset.xml"));
160
	}
161

    
162

    
163
	@Test
164
	public void testParseACM() throws Exception {
165

    
166
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordACM.xml"));
167
	}
168

    
169
	@Test
170
	public void testParseASB() throws Exception {
171

    
172
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("recordASB.xml"));
173
	}
174

    
175
	@Test
176
	public void testParseProjectCorda() throws Exception {
177

    
178
		doTest(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml"));
179
	}
180

    
181
	@Test
182
	public void testParseProjectFCT() throws Exception {
183

    
184
		doTest(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordFCT.xml"));
185
	}
186

    
187

    
188
	@Test
189
	public void testParseOaf() throws Exception {
190

    
191
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("record.xml"));
192
	}
193

    
194
	@Test
195
	public void testParseOafPublication() throws Exception {
196

    
197
		doTest(loadFromTransformationProfile("oaf_entity2hbase.xml"), load("record.xml"));
198
	}
199

    
200
	@Test
201
	public void testParseLindat() throws Exception {
202

    
203
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordLindat.xml"));
204
	}
205

    
206
	@Test
207
	public void testParseDatacite() throws Exception {
208

    
209
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatacite.xml"));
210
	}
211

    
212
	@Test
213
	public void testParseDatacite2() throws Exception {
214

    
215
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("recordDatacite2.xml"));
216
	}
217

    
218
	@Test
219
	public void testParseOpenTrials() throws Exception {
220

    
221
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("opentrials_datacite1.xml"));
222
	}
223

    
224
	@Test
225
	public void testLinkPangaea() throws Exception {
226

    
227
		final List<Row> rows = Lists.newArrayList();
228
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("pangaeODF1.xml")));
229
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("pangaeODF2.xml")));
230
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("pangaeOAF.xml")));
231
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCordaPangaea.xml")));
232

    
233
		printAll(mapAll(buildTable(rows)));
234
	}
235

    
236
	@Test
237
	public void testPangaea() throws Exception {
238

    
239
		final List<Row> rows = Lists.newArrayList();
240
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("pangaeOAF2.xml")));
241
		printAll(mapAll(buildTable(rows)));
242
	}
243
	@Test
244
	public void testZenodo() throws Exception {
245

    
246
		final List<Row> rows = Lists.newArrayList();
247
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("zenodoData.xml")));
248
		printAll(mapAll(buildTable(rows)));
249
	}
250

    
251
	@Test
252
	public void testZenodoSoftware() throws Exception {
253

    
254
		final List<Row> rows = Lists.newArrayList();
255
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("softwareZenodo_odf.xml")));
256
		printAll(mapAll(buildTable(rows)));
257
	}
258

    
259
	@Test
260
	public void testLinkCorda() throws Exception {
261

    
262
		final List<Row> rows = Lists.newArrayList();
263
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml")));
264
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordCorda.xml")));
265

    
266
		printAll(mapAll(buildTable(rows)));
267
	}
268

    
269
	@Test
270
	public void testLinkFCT() throws Exception {
271

    
272
		final List<Row> rows = Lists.newArrayList();
273
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordFCT.xml")));
274
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordFCT.xml")));
275

    
276
		printAll(mapAll(buildTable(rows)));
277
	}
278

    
279
	@Test
280
	public void testLinkARC() throws Exception {
281

    
282
		final List<Row> rows = Lists.newArrayList();
283
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordARC.xml")));
284
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordARC.xml")));
285

    
286
		printAll(mapAll(buildTable(rows)));
287
	}
288

    
289
	@Test
290
	public void testLinkWT() throws Exception {
291

    
292
		final List<Row> rows = Lists.newArrayList();
293
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordWT.xml")));
294
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordWT.xml")));
295

    
296
		printAll(mapAll(buildTable(rows)));
297
	}
298

    
299

    
300

    
301
	@Test
302
	public void testLinkOrganization() throws Exception {
303

    
304
		final List<Row> rows = Lists.newArrayList();
305
		rows.addAll(asRows(loadFromTransformationProfile("organizations_2_hbase.xsl"), load("organization.xml")));
306
		rows.addAll(asRows(loadFromTransformationProfile("projectorganization_2_hbase.xsl"), load("project_organization.xml")));
307
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml")));
308

    
309
		printAll(mapAll(buildTable(rows)));
310
	}
311

    
312
	@Test
313
	public void testLinkOrganizationAffiliation() throws Exception {
314

    
315
		final List<Row> rows = Lists.newArrayList();
316
		rows.addAll(asRows(loadFromTransformationProfile("organizations_2_hbase.xsl"), load("organization.xml")));
317
		rows.addAll(asRows(loadFromTransformationProfile("resultorganization_2_hbase.xsl"), load("result_organization.xml")));
318
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("record.xml")));
319

    
320
		printAll(mapAll(buildTable(rows)));
321
	}
322

    
323
	@Test
324
	public void testDuplicates() throws Exception {
325
		final String mergeId = "50|dedup_wf_001::08ed625d07e5738b794ff14d6773fd9f";
326
		final List<Row> rows = Lists.newArrayList();
327

    
328
		final Function<Row, Row> f = rowIn -> {
329

    
330
			final List<Column<String,byte[]>> cols = Lists.newArrayList();
331
			for(Column<String,byte[]> col : rowIn.getColumns()) {
332
				if (col.getName().equals("body")) {
333
					cols.add(new Column(col.getName(), col.getValue()));
334

    
335
				}
336
			}
337
			return new Row("result", rowIn.getKey(), cols);
338
		};
339

    
340
		final List<Row> puma1 = asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordPuma1.xml"), f);
341
		puma1.add(new Row("resultResult_dedup_isMergedIn", mergeId));
342

    
343
		final List<Row> puma2 = asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordPuma2.xml"), f);
344
		puma2.add(new Row("resultResult_dedup_isMergedIn", mergeId));
345

    
346
		rows.addAll(puma1);
347
		rows.addAll(puma2);
348

    
349
		List<Oaf> duplicates = Lists.newArrayList();
350
		duplicates.add(getOafBody(puma1));
351
		duplicates.add(getOafBody(puma2));
352
		final Oaf.Builder oafMerge = OafEntityMerger.merge(mergeId, duplicates);
353

    
354
		final Row mergeRow = new Row("result", mergeId, Lists.newArrayList(new Column("body", oafMerge.build().toByteArray())));
355

    
356
		rows.add(mergeRow);
357

    
358
		printAll(mapAll(buildTable(rows)));
359
	}
360

    
361
	private Oaf getOafBody(final List<Row> rows) throws InvalidProtocolBufferException {
362
		for(Row row : rows) {
363
			if(StringUtils.startsWith(row.getKey(), "50")) {
364
				return Oaf.parseFrom(row.getColumn("body").getValue());
365

    
366
			}
367
		}
368
		return null;
369
	}
370

    
371
	@Test
372
	public void testParseDoajOAF() throws Exception {
373

    
374
		doTest(loadFromTransformationProfile("oaf2hbase.xml"), load("doajUniversityRecord.xml"));
375
	}
376

    
377
	@Test
378
	public void testParseDatasource() throws Exception {
379

    
380
		doTest(loadFromTransformationProfile("datasources_2_hbase.xsl"), load("datasourceNative.xml"));
381
	}
382
	@Test
383
	public void testParseDatasourcePiwik() throws Exception {
384

    
385
		doTest(loadFromTransformationProfile("datasources_2_hbase.xsl"), load("datasourcePiwik.xml"));
386
	}
387

    
388
	@Test
389
	public void testParseDataDatasource() throws Exception {
390

    
391
		doTest(loadFromTransformationProfile("datasources_2_hbase.xsl"), load("dataDatasource.xml"));
392
	}
393

    
394
	@Test
395
	public void testFromMongodbCompressedDump() throws Exception {
396
		doTestJsonGz(loadFromTransformationProfile("oaf2hbase.xml"), load("mdstore_cleaned.json.gz"));
397
	}
398

    
399

    
400
	@Test
401
	public void testLoadFromTransformationProfile() throws IOException {
402
		InputStream in = loadFromTransformationProfile("oaf2hbase.xml");
403
		log.info(IOUtils.toString(in));
404
	}
405

    
406
	@Test
407
	public void test_template() throws Exception {
408
		final String xslt = IOUtils.toString(loadFromTransformationProfile("oaf2hbase.xml"));
409
		final XsltRowTransformer transformer = factory.getTransformer(xslt);
410
		assertNotNull(transformer);
411

    
412
		final String record = IOUtils.toString(load("record.xml"));
413
		final List<Row> rows = transformer.apply(record);
414

    
415
		System.out.println(rows);
416
	}
417

    
418
	@Test
419
	public void testWrongCharsOrganization() throws Exception {
420
		final List<Row> rows = Lists.newArrayList();
421
		rows.addAll(asRows(loadFromTransformationProfile("organizations_2_hbase.xsl"), load("organizationWrongChars.xml")));
422
		printAll(mapAll(buildTable(rows)));
423
	}
424

    
425
	@Test
426
	public void testParseProjectWithFunderOriginalName() throws Exception {
427

    
428
		doTest(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectWithFunderOriginalName.xml"));
429
	}
430
	@Test
431
	public void testLinkFunderOriginalName() throws Exception {
432

    
433
		final List<Row> rows = Lists.newArrayList();
434
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectWithFunderOriginalName.xml")));
435
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordFunderOriginalName.xml")));
436

    
437
		printAll(mapAll(buildTable(rows)));
438
	}
439

    
440
	@Test
441
	public void testProjectExtraInfo() throws Exception {
442
		final List<Row> rows = Lists.newArrayList();
443
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordExtraInfo.xml")));
444
		printAll(mapAll(buildTable(rows)));
445
	}
446

    
447
	@Test
448
	public void testParseSoftwareFromODF() throws Exception {
449
		final List<Row> rows = Lists.newArrayList();
450
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("softwareODF.xml")));
451
		rows.addAll(asRows(loadFromTransformationProfile("projects_2_hbase.xsl"), load("projectRecordCorda.xml")));
452
		printAll(mapAll(buildTable(rows)));
453
	}
454

    
455
	@Test
456
	public void testParseSoftwareFromOAF() throws Exception {
457
		final List<Row> rows = Lists.newArrayList();
458
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordOAFsoftwareCLOSED.xml")));
459
		printAll(mapAll(buildTable(rows)));
460
	}
461

    
462
	@Test
463
	public void testParsePubFromODF() throws Exception {
464
		final List<Row> rows = Lists.newArrayList();
465
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("recordODFPub.xml")));
466
		printAll(mapAll(buildTable(rows)));
467
	}
468

    
469
	@Test
470
	public void testParseSoftwareFromOAFOpen() throws Exception {
471
		final List<Row> rows = Lists.newArrayList();
472
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("recordOAFsoftwareOPEN.xml")));
473
		printAll(mapAll(buildTable(rows)));
474
	}
475

    
476
	@Test
477
	public void testParseSoftwareBiotool() throws Exception {
478
		final List<Row> rows = Lists.newArrayList();
479
		rows.addAll(asRows(loadFromTransformationProfile("odf2hbase.xml"), load("biotoolSw.xml")));
480
		printAll(mapAll(buildTable(rows)));
481
	}
482

    
483
	@Test
484
	public void testParseOafWithExternalRef() throws Exception {
485
		final List<Row> rows = Lists.newArrayList();
486
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("oafWithExternalReference.xml")));
487
		printAll(mapAll(buildTable(rows)));
488
	}
489

    
490
	@Test
491
	public void testParseOafWithCommunity() throws Exception {
492
		final List<Row> rows = Lists.newArrayList();
493
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("oafWithCommunity.xml")));
494
		printAll(mapAll(buildTable(rows)));
495
	}
496

    
497
	@Test
498
	public void testParseOafWithUpdates() throws Exception {
499
		final List<Row> rows = Lists.newArrayList();
500
		rows.addAll(asRows(loadFromTransformationProfile("oaf2hbase.xml"), load("oafWithCommunity.xml")));
501

    
502
		ActionFactory actionFactory = new ActionFactory();
503

    
504
		Map<String, Resource> xslts = Maps.newHashMap();
505

    
506
		xslts.put("oaf2hbase", new ByteArrayResource(IOUtils.toString(loadFromTransformationProfile("oaf2hbase.xml")).getBytes()));
507
		actionFactory.setXslts(xslts);
508

    
509
		XsltInfoPackageAction pa = actionFactory.generateInfoPackageAction(
510
				"oaf2hbase",
511
				"rawset-id",
512
				new Agent("agent-id", "agent-name", Agent.AGENT_TYPE.algo),
513
				Operation.UPDATE,
514
				IOUtils.toString(load("oafUpdateWithCommunity.xml")),
515
				Provenance.sysimport_mining_aggregator,
516
				"0.9");
517

    
518
		final String qualifier = "update_" + System.nanoTime();
519

    
520
		IOUtils.readLines(load("country_updates.json")).forEach(line -> {
521

    
522
			Oaf.Builder oaf = Oaf.newBuilder();
523

    
524
			try {
525
				JsonFormat.merge(line, oaf);
526
			} catch (JsonFormat.ParseException e) {
527
				throw new IllegalArgumentException(e);
528
			}
529

    
530
			Column<String, byte[]> col = new Column<>("update_" + System.nanoTime(), oaf.build().toByteArray());
531
			rows.add(new Row("result", oaf.getEntity().getId(), Lists.newArrayList(col)));
532
		});
533

    
534
		pa.asAtomicActions().forEach(a -> {
535
			Column<String, byte[]> col = new Column<>("update_" + System.nanoTime(), a.getTargetValue());
536
			rows.add(new Row(a.getTargetColumnFamily(), a.getTargetRowKey(), Lists.newArrayList(col)));
537
		});
538

    
539

    
540
		/*
541
		rows.forEach(r -> {
542
			log.info(r);
543
		});
544
		*/
545

    
546
		mapAll(buildTable(rows)).entrySet().forEach(b -> {
547
			log.info(b.getKey());
548
			log.info(b.getValue());
549
		});
550
	}
551

    
552
	@Test
553
	public void testParseCrisPub() throws Exception {
554
		doTest(loadFromTransformationProfile("odf2hbase.xml"), load("cris_pub1.xml"));
555
	}
556

    
557
	@Test
558
	public void testBioToolSwRowJson() throws Exception {
559
		doTestJsonRow(IOUtils.toString(load("biotoolSwRow.json")));
560
	}
561

    
562
	private void doTestJsonRow(final String json) throws Exception {
563
		Row row = asRowFromJson(json);
564
		log.info(row);
565
		List<Row> rows = new ArrayList<>();
566
		rows.add(row);
567
		final Map<String, Map<String, Map<String, byte[]>>> table = buildTable(rows);
568
		final Map<String, XmlRecordFactory> builders = mapAll(table);
569
		printAll(builders);
570

    
571
	}
572

    
573
	private void doTest(final InputStream xsltStream, final InputStream recordStream) throws Exception {
574
		try {
575
			final List<Row> rows = asRows(xsltStream, recordStream);
576

    
577
			log.info(rows);
578

    
579
			final Map<String, Map<String, Map<String, byte[]>>> table = buildTable(rows);
580

    
581
			// System.out.println("\n" + table.toString());
582

    
583
			final Map<String, XmlRecordFactory> builders = mapAll(table);
584

    
585
			printAll(builders);
586
		} catch (final InvalidProtocolBufferException e) {
587
			throw new Exception(e);
588
		} catch (final TransformerConfigurationException e) {
589
			throw new Exception(e);
590
		} catch (final TransformerFactoryConfigurationError e) {
591
			throw new Exception(e);
592
		} catch (final DocumentException e) {
593
			throw new Exception(e);
594
		}
595
	}
596

    
597
	private void doTestJsonGz(final InputStream xsltStream, final InputStream recordStream) throws Exception {
598

    
599
		final Iterator<List<Row>> rowsIterator = asRowsJsonGzip(xsltStream, recordStream);
600

    
601
		int i = 0;
602
		while (rowsIterator.hasNext()) {
603
			final List<Row> rows = rowsIterator.next();
604
			i++;
605

    
606
			if ((i % 10000) == 0) {
607
				System.out.println(i);
608
			}
609

    
610
			final Map<String, Map<String, Map<String, byte[]>>> table = buildTableDoaj(rows);
611

    
612
			for (final Map<String, Map<String, byte[]>> m : table.values()) {
613
				for (final Map<String, byte[]> mv : m.values()) {
614
					for (final byte[] v : mv.values()) {
615
						final OafDecoder d = OafDecoder.decode(v);
616
						assertNotNull(d);
617
						assertNotNull(d.getOaf());
618

    
619
						switch (d.getKind()) {
620
						case entity:
621
							assertNotNull(d.getMetadata());
622
							if (d.getOaf().getEntity().getType().equals(Type.result)) {
623
								System.out.println(d.getOaf());
624
							}
625
							break;
626
						case relation:
627
							assertNotNull(d.getRel());
628
							break;
629
						default:
630
							break;
631
						}
632
					}
633
				}
634
			}
635
		}
636
	}
637

    
638
	private List<Row> asRows(final InputStream xsltStream, final InputStream recordStream, final Function<Row, Row> p) throws Exception {
639
		return asRows(xsltStream, new HashMap<>(), recordStream, p);
640
	}
641

    
642
	private List<Row> asRows(final InputStream xsltStream, final InputStream recordStream) throws Exception {
643
		return asRows(xsltStream, new HashMap<>(), recordStream);
644
	}
645

    
646
	private List<Row> asRows(final InputStream xsltStream, final Map<String, Object> params, final InputStream recordStream) throws Exception {
647
		return asRows(xsltStream, params, recordStream, null);
648
	}
649

    
650
	private List<Row> asRows(final InputStream xsltStream, final Map<String, Object> params, final InputStream recordStream, final Function<Row, Row> p) throws Exception {
651
		final String xslt = IOUtils.toString(xsltStream);
652
		final XsltRowTransformer transformer = factory.getTransformer(xslt, params);
653
		assertNotNull(transformer);
654

    
655
		final String record = IOUtils.toString(recordStream);
656
		final List<Row> rows = transformer.apply(record);
657

    
658
		assertNotNull(rows);
659
		assertFalse(rows.isEmpty());
660
		return p == null ? rows : Lists.newArrayList(Iterables.transform(rows, p));
661
	}
662

    
663
	private Iterator<List<Row>> asRowsJsonGzip(final InputStream xsltStream, final InputStream recordStreamJsonGzip) throws Exception {
664
		final String xslt = IOUtils.toString(xsltStream);
665
		final XsltRowTransformer transformer = factory.getTransformer(xslt);
666
		assertNotNull(transformer);
667
		assertNotNull(recordStreamJsonGzip);
668

    
669
		final GZIPInputStream stream = new GZIPInputStream(recordStreamJsonGzip);
670
		assertNotNull(stream);
671
		final BufferedReader inStream = new BufferedReader(new InputStreamReader(stream));
672
		assertNotNull(inStream);
673
		return new Iterator<List<Row>>() {
674

    
675
			String jsonRecord = null;
676

    
677
			@Override
678
			public boolean hasNext() {
679
				try {
680
					return (jsonRecord = inStream.readLine()) != null;
681
				} catch (final IOException e) {
682
					throw new RuntimeException(e);
683
				}
684
			}
685

    
686
			@Override
687
			public List<Row> next() {
688

    
689
				final JSONObject jsonObj = new JSONObject(jsonRecord);
690
				final String body = jsonObj.getString("body");
691
				try {
692
					assertTrue(StringUtils.isNotBlank(body));
693
					// System.out.println(body);
694
					final List<Row> rows = transformer.apply(body);
695
					assertNotNull(rows);
696
					assertFalse(rows.isEmpty());
697
					return rows;
698
				} catch (final Throwable e) {
699
					System.err.println("error transforming document: " + body);
700
					throw new RuntimeException(e);
701
				}
702
			}
703

    
704
			@Override
705
			public void remove() {
706
				throw new UnsupportedOperationException();
707
			}
708

    
709
		};
710

    
711
	}
712

    
713
	private Map<String, Map<String, Map<String, byte[]>>> buildTableDoaj(final List<Row> rows) throws UnsupportedEncodingException {
714
		final Map<String, Map<String, Map<String, byte[]>>> table = Maps.newHashMap();
715

    
716
		for (final Row row : rows) {
717
			final String rowKey = row.getKey();
718
			final String cf = row.getColumnFamily();
719
			if (!table.containsKey(rowKey)) {
720
				table.put(rowKey, new HashMap<>());
721
			}
722
			if (!table.get(rowKey).containsKey(cf)) {
723
				table.get(rowKey).put(row.getColumnFamily(), new HashMap<>());
724
			}
725
			for (final Column<String, byte[]> c : row.getColumns()) {
726
				// System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));
727
				table.get(rowKey).get(cf).put(c.getName(), c.getValue());
728
				if (cf.equals("result") && c.getName().equals("body")) {
729
					// System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));
730
					assertTrue(StringUtils.isNotBlank(new String(c.getValue(), "UTF-8")));
731
				}
732
			}
733
		}
734
		return table;
735

    
736
	}
737

    
738
	protected Map<String, Map<String, Map<String, byte[]>>> buildTable(final List<Row> rows) throws UnsupportedEncodingException {
739
		final Map<String, Map<String, Map<String, byte[]>>> table = Maps.newHashMap();
740

    
741
		for (final Row row : rows) {
742
			final String rowKey = row.getKey();
743
			final String cf = row.getColumnFamily();
744
			if (!table.containsKey(rowKey)) {
745
				table.put(rowKey, new HashMap<>());
746
			}
747
			if (!table.get(rowKey).containsKey(cf)) {
748
				table.get(rowKey).put(row.getColumnFamily(), new HashMap<>());
749
			}
750
			for (final Column<String, byte[]> c : row.getColumns()) {
751
				System.out.println(String.format("ADDING K:%s CF:%s Q:%s", rowKey, cf, c.getName()));
752
				table.get(rowKey).get(cf).put(c.getName(), c.getValue());
753
				if (c.getName().equals("body")) {
754
					final String theBody = new String(c.getValue(), "UTF-8");
755
					assertTrue(StringUtils.isNotBlank(theBody));
756
					//System.out.println(theBody);
757
				}
758
			}
759
		}
760
		return table;
761

    
762
	}
763

    
764
	protected Map<String, XmlRecordFactory> mapAll(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {
765

    
766
		final Map<String, XmlRecordFactory> builders = Maps.newHashMap();
767
		for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {
768
			map(builders, e.getKey(), e.getValue());
769
		}
770
		return builders;
771
	}
772

    
773
	// private Map<String, XmlRecordFactory> mapResultsOnly(final Map<String, Map<String, Map<String, byte[]>>> table) throws Exception {
774
	//
775
	// final Map<String, XmlRecordFactory> builders = Maps.newHashMap();
776
	// for (final Entry<String, Map<String, Map<String, byte[]>>> e : table.entrySet()) {
777
	// final Type type = OafRowKeyDecoder.decode(e.getKey()).getType();
778
	// if (type == Type.result) {
779
	// map(builders, e.getKey(), e.getValue());
780
	// }
781
	// }
782
	// return builders;
783
	// }
784

    
785
	private void map(final Map<String, XmlRecordFactory> builders, final String rowKey, final Map<String, Map<String, byte[]>> row) throws Exception {
786

    
787
		final Type type = OafRowKeyDecoder.decode(rowKey).getType();
788

    
789
		final Map<String, byte[]> familyMap = row.get(type.toString());
790

    
791
		if (familyMap == null) return;
792

    
793
		final byte[] bodyB = familyMap.get("body");
794

    
795
		if (bodyB != null) {
796
			ensureBuilder(builders, rowKey);
797

    
798
			final Oaf oaf = UpdateMerger.mergeBodyUpdates(familyMap);
799

    
800
			final OafDecoder mainEntity = OafDecoder.decode(oaf);
801

    
802
			builders.get(rowKey).setMainEntity(mainEntity);
803

    
804
			for (final LinkDescriptor ld : entityConfigTable.getDescriptors(type)) {
805

    
806
				final String it = ld.getRelDescriptor().getIt();
807
				final Map<String, byte[]> cols = row.get(it);
808

    
809
				if ((cols != null) && !cols.isEmpty()) {
810

    
811
					for (final byte[] oafB : cols.values()) {
812

    
813
						final Oaf.Builder relBuilder = Oaf.newBuilder(Oaf.parseFrom(oafB));
814

    
815
						if (ld.isSymmetric()) {
816
							final RelDescriptor rd = ld.getRelDescriptor();
817

    
818
							relBuilder.getRelBuilder().setCachedTarget(mainEntity.getEntity()).setRelType(rd.getRelType()).setSubRelType(rd.getSubRelType());
819
						}
820

    
821
						relBuilder.getRelBuilder().setChild(ld.isChild());
822

    
823
						final Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(Kind.relation).setLastupdatetimestamp(System.currentTimeMillis());
824
						oafBuilder.mergeFrom(relBuilder.build());
825

    
826
						final String targetId = ld.isSymmetric() ? oafBuilder.getRel().getTarget() : oafBuilder.getRel().getSource();
827
						ensureBuilder(builders, targetId);
828
						final OafDecoder decoder = OafDecoder.decode(oafBuilder.build());
829

    
830
						if (ld.isChild()) {
831
							builders.get(targetId).addChild(type, decoder);
832
						} else {
833
							builders.get(targetId).addRelation(type, decoder);
834
						}
835
					}
836

    
837
				}
838
			}
839
		}
840

    
841
	}
842

    
843
	private void printAll(final Map<String, XmlRecordFactory> builders) throws DocumentException {
844
		print(Sets.newHashSet(Type.values()), builders, null);
845
	}
846

    
847
	private void print(final Set<Type> types, final Map<String, XmlRecordFactory> builders, final Map<Type, Set<String>> xpaths) throws DocumentException {
848
		final SAXReader r = new SAXReader();
849

    
850
		for (final Entry<String, XmlRecordFactory> e : builders.entrySet()) {
851
			final OafRowKeyDecoder kd = OafRowKeyDecoder.decode(e.getKey());
852

    
853
			if (!e.getValue().isValid()) throw new IllegalArgumentException("invalid builder: " + e.getKey());
854
			if (types.contains(kd.getType())) {
855
				final String val = IndentXmlString.apply(e.getValue().build());
856

    
857
				if ((xpaths != null) && !xpaths.isEmpty() && (xpaths.get(kd.getType()) != null)) {
858
					final Document doc = r.read(new StringReader(val));
859

    
860
					log.debug("\n" + e.getKey());
861
					for (final String xpath : xpaths.get(kd.getType())) {
862
						log.debug(doc.valueOf(xpath));
863
					}
864
				} else {
865
					log.info(val);
866
				}
867
			}
868
		}
869
	}
870

    
871
	private void printNoIndent(final Map<String, XmlRecordFactory> builders) {
872
		for (final Entry<String, XmlRecordFactory> e : builders.entrySet()) {
873
			if (e.getValue().isValid()) {
874
				log.debug(e.getValue().build());
875
			} else {
876
				log.debug("invalid builder: " + e.getKey());
877
			}
878
		}
879
	}
880

    
881
	private void ensureBuilder(final Map<String, XmlRecordFactory> builders, final String rowKey) throws Exception {
882
		if (!builders.containsKey(rowKey)) {
883
			builders.put(rowKey, newBuilder());
884
		}
885
	}
886

    
887
	private XmlRecordFactory newBuilder() throws TransformerConfigurationException, TransformerFactoryConfigurationError, DocumentException {
888
		return new XmlRecordFactory(entityConfigTable, ContextMapper.fromXml(Context.xml),
889
				RelClasses.fromJSon(RelClassesTest.relClassesJson), XmlRecordFactoryTest.SCHEMA_LOCATION, true, false, false, XmlRecordFactoryTest.specialDatasourceTypes);
890
	}
891

    
892
	private InputStream load(final String fileName) {
893
		return getClass().getResourceAsStream(fileName);
894
	}
895

    
896
	private InputStream loadFromTransformationProfile(final String profilePath) {
897
		log.info("Loading xslt from: " + basePathProfiles + profilePath);
898
		InputStream profile = getClass().getResourceAsStream(basePathProfiles + profilePath);
899
		SAXReader saxReader = new SAXReader();
900
		Document doc = null;
901
		try {
902
			doc = saxReader.read(profile);
903
		} catch (DocumentException e) {
904
			e.printStackTrace();
905
			throw new RuntimeException(e);
906
		}
907
		String xslt = doc.selectSingleNode("//SCRIPT/CODE/*[local-name()='stylesheet']").asXML();
908
		//log.info(xslt);
909
		return IOUtils.toInputStream(xslt);
910
	}
911

    
912

    
913
	private Row asRowFromJson(String json) throws ParseException {
914
		Oaf.Builder oafBuilder = Oaf.newBuilder();
915
		JsonFormat.merge(json, oafBuilder);
916
		final Oaf oaf = oafBuilder.build();
917
		Row row = new Row("result", oaf.getEntity().getId());
918
		Column<String, byte[]> c = new Column<>("body", oaf.toByteArray());
919
		row.setColumn("body", c);
920
		return row;
921

    
922
	}
923

    
924
}
    (1-1/1)