1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.docproc.util;
import com.yahoo.document.DocumentOperation;
import com.yahoo.document.DocumentPut;
import com.yahoo.document.config.DocumentmanagerConfig;
import com.yahoo.config.docproc.SplitterJoinerDocumentProcessorConfig;
import com.yahoo.docproc.DocumentProcessor;
import com.yahoo.docproc.Processing;
import com.yahoo.document.ArrayDataType;
import com.yahoo.document.Document;
import com.yahoo.document.DocumentType;
import com.yahoo.document.DocumentTypeManager;
import com.yahoo.document.DocumentTypeManagerConfigurer;
import com.yahoo.document.datatypes.Array;
import com.yahoo.log.LogLevel;
import java.util.logging.Logger;
import java.util.stream.Collectors;
/**
* @author <a href="mailto:einarmr@yahoo-inc.com">Einar M R Rosenvinge</a>
*/
public class SplitterDocumentProcessor extends DocumentProcessor {
private static Logger log = Logger.getLogger(SplitterDocumentProcessor.class.getName());
private String documentTypeName;
private String arrayFieldName;
private String contextFieldName;
DocumentTypeManager manager;
public SplitterDocumentProcessor(SplitterJoinerDocumentProcessorConfig cfg, DocumentmanagerConfig documentmanagerConfig) {
super();
this.documentTypeName = cfg.documentTypeName();
this.arrayFieldName = cfg.arrayFieldName();
this.contextFieldName = cfg.contextFieldName();
this.manager = DocumentTypeManagerConfigurer.configureNewManager(documentmanagerConfig);
validate(manager, documentTypeName, arrayFieldName);
}
@Override
public Progress process(Processing processing) {
if (processing.getDocumentOperations().size() != 1) {
//we were given more than one document, return
log.log(LogLevel.DEBUG, "More than one document given, returning. (Was given "
+ processing.getDocumentOperations().size() + " documents).");
return Progress.DONE;
}
if (!doProcessOuterDocument(processing.getDocumentOperations().get(0), documentTypeName)) {
return Progress.DONE;
}
Document outerDoc = ((DocumentPut)processing.getDocumentOperations().get(0)).getDocument();;
@SuppressWarnings("unchecked")
Array<Document> innerDocuments = (Array<Document>) outerDoc.getFieldValue(arrayFieldName);
if (innerDocuments == null) {
//the document does not have the field, return
log.log(LogLevel.DEBUG, "The given Document does not have a field value for field "
+ arrayFieldName + ", returning. (Was given " + outerDoc + ").");
return Progress.DONE;
}
if (innerDocuments.size() == 0) {
//the array is empty, return
log.log(LogLevel.DEBUG, "The given Document does not have any elements in array field "
+ arrayFieldName + ", returning. (Was given " + outerDoc + ").");
return Progress.DONE;
}
split(processing, innerDocuments);
return Progress.DONE;
}
private void split(Processing processing, Array<Document> innerDocuments) {
processing.setVariable(contextFieldName, processing.getDocumentOperations().get(0));
processing.getDocumentOperations().clear();
processing.getDocumentOperations().addAll(innerDocuments.stream()
.map(DocumentPut::new)
.collect(Collectors.toList()));
innerDocuments.clear();
}
static void validate(DocumentTypeManager manager, String documentTypeName, String arrayFieldName) {
DocumentType docType = manager.getDocumentType(documentTypeName);
if (docType == null) {
//the document type does not exist, return
throw new IllegalStateException("The document type " + documentTypeName + " is not deployed.");
}
if (docType.getField(arrayFieldName) == null) {
//the document type does not have the field, return
throw new IllegalStateException("The document type " + documentTypeName
+ " does not have a field named " + arrayFieldName + ".");
}
if (!(docType.getField(arrayFieldName).getDataType() instanceof ArrayDataType)) {
//the data type of the field is wrong, return
throw new IllegalStateException("The data type of the field named "
+ arrayFieldName + " in document type " + documentTypeName
+ " is not an array type");
}
ArrayDataType fieldDataType = (ArrayDataType) docType.getField(arrayFieldName).getDataType();
if (!(fieldDataType.getNestedType() instanceof DocumentType)) {
//the subtype of tye array data type of the field is wrong, return
throw new IllegalStateException("The data type of the field named "
+ arrayFieldName + " in document type " + documentTypeName
+ " is not an array of Document.");
}
}
static boolean doProcessOuterDocument(Object o, String documentTypeName) {
if ( ! (o instanceof DocumentOperation)) {
if (log.isLoggable(LogLevel.DEBUG)) {
log.log(LogLevel.DEBUG, o + " is not a DocumentOperation.");
}
return false;
}
DocumentOperation outerDocOp = (DocumentOperation)o;
if ( ! (outerDocOp instanceof DocumentPut)) {
//this is not a put, return
if (log.isLoggable(LogLevel.DEBUG)) {
log.log(LogLevel.DEBUG, "Given DocumentOperation is not a DocumentPut, returning. (Was given "
+ outerDocOp + ").");
}
return false;
}
Document outerDoc = ((DocumentPut) outerDocOp).getDocument();
DocumentType type = outerDoc.getDataType();
if (!type.getName().equalsIgnoreCase(documentTypeName)) {
//this is not the right document type
if (log.isLoggable(LogLevel.DEBUG)) {
log.log(LogLevel.DEBUG, "Given Document is of wrong type, returning. (Was given " + outerDoc + ").");
}
return false;
}
return true;
}
}
|
