Drop KvinTupleInternal and add KvinRecord-based writer.

Author: kenwenzel

bundles/io.github.linkedfactory.core/src/main/java/io/github/linkedfactory/core/kvin/parquet/Compactor.java

@@ -1,5 +1,7 @@
 package io.github.linkedfactory.core.kvin.parquet;
 
+import io.github.linkedfactory.core.kvin.parquet.records.KvinRecord;
+import net.enilink.commons.iterator.IExtendedIterator;
 import net.enilink.commons.util.Pair;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -153,13 +155,6 @@ private ParquetReader<IdMapping> getParquetMappingReader(HadoopInputFile file) t
 				.build();
 	}
 
-	private ParquetReader<KvinTupleInternal> getParquetDataReader(HadoopInputFile file) throws IOException {
-		return AvroParquetReader.<KvinTupleInternal>builder(file)
-				.withDataModel(reflectData)
-				.useStatsFilter()
-				.build();
-	}
-
 	private void compactDataFiles(File weekFolder) throws IOException {
 		Lock readLock = kvinParquet.readLock();
 		try {
@@ -182,49 +177,34 @@ private void compactDataFiles(File weekFolder) throws IOException {
 					Paths.get(archiveLocation).relativize(weekFolder.toPath()));
 
 			Path compactionFile = new Path(targetFolder.toAbsolutePath().toString(), "data__1.parquet");
-			ParquetWriter<KvinTupleInternal> compactionFileWriter = getParquetDataWriter(compactionFile);
+			ParquetWriter<KvinRecord> compactionFileWriter = getKvinRecordWriter(compactionFile);
 
-			PriorityQueue<Pair<KvinTupleInternal, ParquetReader<KvinTupleInternal>>> nextTuples =
+			PriorityQueue<Pair<KvinRecord, IExtendedIterator<KvinRecord>>> nextRecords =
 					new PriorityQueue<>(Comparator.comparing(Pair::getFirst));
 			for (java.nio.file.Path dataFile : dataFiles) {
-				ParquetReader<KvinTupleInternal> reader = getParquetDataReader(
-						HadoopInputFile.fromPath(new Path(dataFile.toString()), new Configuration()));
-				KvinTupleInternal tuple = reader.read();
-				if (tuple != null) {
-					nextTuples.add(new Pair<>(tuple, reader));
+				IExtendedIterator<KvinRecord> it = createKvinRecordReader(new Path(dataFile.toString()), null);
+				if (it.hasNext()) {
+					nextRecords.add(new Pair<>(it.next(), it));
 				} else {
-					try {
-						reader.close();
-					} catch (IOException e) {
-					}
+					it.close();
 				}
 			}
 
-			KvinTupleInternal prevTuple = null;
-			while (!nextTuples.isEmpty()) {
-				var pair = nextTuples.poll();
-				if (prevTuple == null || prevTuple.compareTo(pair.getFirst()) != 0) {
+			KvinRecord prevRecord = null;
+			while (!nextRecords.isEmpty()) {
+				var pair = nextRecords.poll();
+				if (prevRecord == null || prevRecord.compareTo(pair.getFirst()) != 0) {
 					var tuple = pair.getFirst();
-					// update the first flag
-					if (prevTuple == null || !Arrays.equals(prevTuple.id, tuple.id)) {
-						tuple.setFirst(true);
-					} else {
-						tuple.setFirst(null);
-					}
 					compactionFileWriter.write(tuple);
-					prevTuple = tuple;
-				} else if (prevTuple != null) {
+					prevRecord = tuple;
+				} else if (prevRecord != null) {
 					// omit tuple as it is duplicate in terms of id, time, and seqNr
 				}
 
-				KvinTupleInternal tuple = pair.getSecond().read();
-				if (tuple != null) {
-					nextTuples.add(new Pair<>(tuple, pair.getSecond()));
+				if (pair.getSecond().hasNext()) {
+					nextRecords.add(new Pair<>(pair.getSecond().next(), pair.getSecond()));
 				} else {
-					try {
-						pair.getSecond().close();
-					} catch (IOException e) {
-					}
+					pair.getSecond().close();
 				}
 			}
 

bundles/io.github.linkedfactory.core/src/main/java/io/github/linkedfactory/core/kvin/parquet/KvinParquet.java

@@ -275,9 +275,8 @@ private synchronized void putInternal(Iterable<KvinTuple> tuples) throws IOExcep
 
 			WriterState writerState = null;
 			String prevKey = null;
-			KvinTupleInternal prevTuple = null;
 			for (KvinTuple tuple : tuples) {
-				KvinTupleInternal internalTuple = new KvinTupleInternal();
+				KvinRecord record = new KvinRecord();
 
 				Calendar tupleDate = getDate(tuple.time);
 				int year = tupleDate.get(Calendar.YEAR);
@@ -293,39 +292,32 @@ private synchronized void putInternal(Iterable<KvinTuple> tuples) throws IOExcep
 								.resolve(weekFolderName)
 								.resolve("data__1.parquet");
 						Files.createDirectories(file.getParent());
-						writerState = new WriterState(file, getParquetDataWriter(new Path(file.toString())),
+						writerState = new WriterState(file, getKvinRecordWriter(new Path(file.toString())),
 								year, week);
 						writers.put(key, writerState);
 					}
 					prevKey = key;
 				}
 
 				// writing mappings and values
-				internalTuple.setId(generateId(tuple, writeContext,
-						itemMappingWriter, propertyMappingWriter, contextMappingWriter));
-				internalTuple.setTime(tuple.time);
-				internalTuple.setSeqNr(tuple.seqNr);
-
-				internalTuple.setValueInt(tuple.value instanceof Integer ? (int) tuple.value : null);
-				internalTuple.setValueLong(tuple.value instanceof Long ? (long) tuple.value : null);
-				internalTuple.setValueFloat(tuple.value instanceof Float ? (float) tuple.value : null);
-				internalTuple.setValueDouble(tuple.value instanceof Double ? (double) tuple.value : null);
-				internalTuple.setValueString(tuple.value instanceof String ? (String) tuple.value : null);
-				internalTuple.setValueBool(tuple.value instanceof Boolean ? (Boolean) tuple.value ? 1 : 0 : null);
-				if (tuple.value instanceof Record || tuple.value instanceof URI || tuple.value instanceof BigInteger ||
-						tuple.value instanceof BigDecimal || tuple.value instanceof Short || tuple.value instanceof Object[]) {
-					internalTuple.setValueObject(encodeRecord(tuple.value));
-				} else {
-					internalTuple.setValueObject(null);
-				}
-				// set first flag
-				if (prevTuple == null || !Arrays.equals(prevTuple.id, internalTuple.id)) {
-					internalTuple.setFirst(true);
+				long[] id = generateIds(tuple, writeContext,
+						itemMappingWriter, propertyMappingWriter, contextMappingWriter);
+				record.itemId = id[0];
+				record.contextId = id[1];
+				record.propertyId = id[2];
+				record.time = tuple.time;
+				record.seqNr = tuple.seqNr;
+
+				Object value = tuple.value;
+				if (value instanceof Record || value instanceof URI || value instanceof BigInteger ||
+						value instanceof BigDecimal || value instanceof Short || value instanceof Object[]) {
+					value = ByteBuffer.wrap(encodeRecord(value));
 				}
-				writerState.writer.write(internalTuple);
+				record.value = value;
+
+				writerState.writer.write(record);
 				writerState.minMax[0] = Math.min(writerState.minMax[0], writeContext.lastItemId);
 				writerState.minMax[1] = Math.max(writerState.minMax[1], writeContext.lastItemId);
-				prevTuple = internalTuple;
 			}
 
 			for (WriterState state : writers.values()) {
@@ -558,7 +550,7 @@ private Calendar getDate(long timestamp) {
 		return calendar;
 	}
 
-	private byte[] generateId(KvinTuple tuple,
+	private long[] generateIds(KvinTuple tuple,
 	                          WriteContext writeContext,
 	                          ParquetWriter itemMappingWriter,
 	                          ParquetWriter propertyMappingWriter,
@@ -620,12 +612,7 @@ private byte[] generateId(KvinTuple tuple,
 			}
 			return newId;
 		});
-
-		ByteBuffer idBuffer = ByteBuffer.allocate(Long.BYTES * 3);
-		idBuffer.putLong(itemId);
-		idBuffer.putLong(contextId);
-		idBuffer.putLong(propertyId);
-		return idBuffer.array();
+		return new long[] {itemId, contextId, propertyId};
 	}
 
 	private long getId(URI entity, IdType idType) {
@@ -1445,12 +1432,12 @@ static class InputFileInfo {
 
 	static class WriterState {
 		java.nio.file.Path file;
-		ParquetWriter<KvinTupleInternal> writer;
+		ParquetWriter<KvinRecord> writer;
 		int year;
 		int week;
 		long[] minMax = {Long.MAX_VALUE, Long.MIN_VALUE};
 
-		WriterState(java.nio.file.Path file, ParquetWriter<KvinTupleInternal> writer, int year, int week) {
+		WriterState(java.nio.file.Path file, ParquetWriter<KvinRecord> writer, int year, int week) {
 			this.file = file;
 			this.writer = writer;
 			this.year = year;