Skip to content

Commit a0e9df9

Browse files
hansvahansva
authored
Merge pull request #3314 from bamaer/3281
Apache POI upgrade, Excel writer minor changes, Redshift bulk loader fields list #3281 #3310
2 parents 06d2b69 + 74bb816 commit a0e9df9

8 files changed

Lines changed: 226 additions & 211 deletions

File tree

‎assemblies/plugins/transforms/excel/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@
3535
<description />
3636

3737
<properties>
38-
<poi.version>5.2.3</poi.version>
38+
<poi.version>5.2.4</poi.version>
3939
<xerces.version>2.12.2</xerces.version>
4040
<odfdom-java.version>0.8.7</odfdom-java.version>
4141
</properties>

‎assemblies/plugins/transforms/formula/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@
3131
<name>Hop Assemblies Plugins Transforms Formula</name>
3232

3333
<properties>
34-
<poi.version>5.2.3</poi.version>
34+
<poi.version>5.2.4</poi.version>
3535
</properties>
3636

3737
<description />

‎core/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@
4444
<commons-collections4.version>4.4</commons-collections4.version>
4545
<commons-lang.version>2.6</commons-lang.version>
4646
<commons-vfs2.version>2.9.0</commons-vfs2.version>
47-
<commons-io.version>2.11.0</commons-io.version>
47+
<commons-io.version>2.14.0</commons-io.version>
4848
<commons-dbcp.version>1.4</commons-dbcp.version>
4949
<commons-pool.version>1.5.7</commons-pool.version>
5050
<commons-net.version>3.9.0</commons-net.version>

‎plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoader.java‎

Lines changed: 51 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,7 @@
2828
import org.apache.hop.core.row.IRowMeta;
2929
import org.apache.hop.core.row.IValueMeta;
3030
import org.apache.hop.core.row.RowMeta;
31-
import org.apache.hop.core.row.value.ValueMetaBigNumber;
3231
import org.apache.hop.core.row.value.ValueMetaDate;
33-
import org.apache.hop.core.row.value.ValueMetaString;
3432
import org.apache.hop.core.util.Utils;
3533
import org.apache.hop.core.vfs.HopVfs;
3634
import org.apache.hop.i18n.BaseMessages;
@@ -80,11 +78,12 @@ public boolean init() {
8078

8179
if(meta.isStreamToS3Csv()){
8280
// get the file output stream to write to S3
83-
data.writer = HopVfs.getOutputStream(meta.getCopyFromFilename(), false);
81+
data.writer = HopVfs.getOutputStream(resolve(meta.getCopyFromFilename()), false);
8482
}
8583

8684
data.db = new Database(this, this, data.databaseMeta);
8785
data.db.connect();
86+
getDbFields();
8887

8988
if (log.isBasic()) {
9089
logBasic(BaseMessages.getString(PKG, "RedshiftBulkLoader.Connection.Connected", data.db.getDatabaseMeta()));
@@ -124,8 +123,14 @@ public boolean processRow() throws HopException {
124123
stmt.close();
125124
conn.close();
126125
}catch(SQLException sqle){
126+
setErrors(1);
127+
stopAll();
128+
setOutputDone(); // signal end to receiver(s)
127129
throw new HopDatabaseException("Error executing COPY statements", sqle);
128130
} catch (IOException ioe) {
131+
setErrors(1);
132+
stopAll();
133+
setOutputDone(); // signal end to receiver(s)
129134
throw new HopTransformException("Error releasing resources", ioe);
130135
}
131136
return false;
@@ -134,6 +139,7 @@ public boolean processRow() throws HopException {
134139
if (first && meta.isStreamToS3Csv()) {
135140

136141
first = false;
142+
data.fieldnrs = new HashMap<>();
137143

138144
if (meta.isTruncateTable()) {
139145
truncateTable();
@@ -145,13 +151,13 @@ public boolean processRow() throws HopException {
145151
if(meta.isStreamToS3Csv()){
146152

147153
}
154+
// write all fields in the stream to Redshift
148155
if (!meta.specifyFields()){
149156

150157
// Just take the whole input row
151158
data.insertRowMeta = getInputRowMeta().clone();
152159
data.selectedRowFieldIndices = new int[data.insertRowMeta.size()];
153160

154-
data.fieldnrs = new HashMap<>();
155161
try{
156162
getDbFields();
157163
}catch(HopException e){
@@ -196,13 +202,14 @@ public boolean processRow() throws HopException {
196202

197203
} else {
198204

205+
// use the columns/fields mapping.
199206
int numberOfInsertFields = meta.getFields().size();
200207
data.insertRowMeta = new RowMeta();
201208

202209
// Cache the position of the selected fields in the row array
203210
data.selectedRowFieldIndices = new int[numberOfInsertFields];
204-
for (int insertFieldIdx = 0; insertFieldIdx < numberOfInsertFields; insertFieldIdx++) {
205-
RedshiftBulkLoaderField vbf = meta.getFields().get(insertFieldIdx);
211+
for(int i=0; i < data.dbFields.size(); i++){
212+
RedshiftBulkLoaderField vbf = meta.getFields().get(i);
206213
String inputFieldName = vbf.getStreamField();
207214
int inputFieldIdx = getInputRowMeta().indexOfValue(inputFieldName);
208215
if (inputFieldIdx < 0) {
@@ -212,7 +219,7 @@ public boolean processRow() throws HopException {
212219
"RedshiftBulkLoader.Exception.FieldRequired",
213220
inputFieldName)); //$NON-NLS-1$
214221
}
215-
data.selectedRowFieldIndices[insertFieldIdx] = inputFieldIdx;
222+
data.selectedRowFieldIndices[i] = inputFieldIdx;
216223

217224
String insertFieldName = vbf.getDatabaseField();
218225
IValueMeta inputValueMeta = getInputRowMeta().getValueMeta(inputFieldIdx);
@@ -226,12 +233,13 @@ public boolean processRow() throws HopException {
226233
IValueMeta insertValueMeta = inputValueMeta.clone();
227234
insertValueMeta.setName(insertFieldName);
228235
data.insertRowMeta.addValueMeta(insertValueMeta);
236+
data.fieldnrs.put(meta.getFields().get(i).getDatabaseField().toUpperCase(), inputFieldIdx);
229237
}
230238
}
231239
}
232240

233241
if(meta.isStreamToS3Csv()){
234-
writeRowToFile(data.outputRowMeta, r);
242+
writeRowToFile(data.insertRowMeta, r);
235243
putRow(data.outputRowMeta, r);
236244
}
237245

@@ -279,7 +287,7 @@ private String buildCopyStatementSqlString() {
279287

280288
if(meta.isStreamToS3Csv() || meta.getLoadFromExistingFileFormat().equals("CSV")){
281289
sb.append(" (");
282-
final IRowMeta fields = data.insertRowMeta;
290+
final IRowMeta fields = data.outputRowMeta;
283291
for (int i = 0; i < fields.size(); i++) {
284292
if (i > 0) {
285293
sb.append(", " + fields.getValueMeta(i).getName());
@@ -291,6 +299,8 @@ private String buildCopyStatementSqlString() {
291299
}
292300

293301
sb.append(" FROM '" + resolve(meta.getCopyFromFilename()) + "'");
302+
sb.append(" NULL '' ");
303+
sb.append(" EMPTYASNULL ");
294304
if(meta.isStreamToS3Csv() || meta.getLoadFromExistingFileFormat().equals("CSV")){
295305
sb.append(" delimiter ','");
296306
}
@@ -308,7 +318,7 @@ private String buildCopyStatementSqlString() {
308318
}
309319
sb.append(" CREDENTIALS 'aws_access_key_id=" + awsAccessKeyId + ";aws_secret_access_key=" + awsSecretAccessKey + "'");
310320
}
311-
if(meta.getLoadFromExistingFileFormat().equals("Parquet")){
321+
if(!StringUtils.isEmpty(meta.getLoadFromExistingFileFormat()) && meta.getLoadFromExistingFileFormat().equals("Parquet")){
312322
sb.append(" FORMAT AS PARQUET;");
313323
}
314324

@@ -343,11 +353,9 @@ private Object[] writeToOutputStream(Object[] r) throws HopException, IOExceptio
343353
*/
344354
private void getDbFields() throws HopException {
345355
data.dbFields = new ArrayList<>();
346-
String sql = "desc table ";
347356

348357
IRowMeta rowMeta = null;
349358

350-
351359
if (!StringUtils.isEmpty(resolve(meta.getSchemaName()))) {
352360
rowMeta = data.db.getTableFields(meta.getSchemaName() + "." + meta.getTableName());
353361
}else {
@@ -432,47 +440,52 @@ private void writeRowToFile(IRowMeta rowMeta, Object[] row) throws HopTransformE
432440
writeField(v, valueData, null);
433441
}
434442
data.writer.write(data.binaryNewline);
435-
} else if (meta.isStreamToS3Csv()) {
443+
} else if (meta.isStreamToS3Csv() && meta.isSpecifyFields()) {
436444
/*
437445
* Only write the fields specified!
438446
*/
439-
for (int i = 0; i < data.dbFields.size(); i++) {
440-
if (data.dbFields.get(i) != null) {
447+
for(int i=0; i < meta.getFields().size(); i++){
448+
if(meta.getFields().get(i).getDatabaseField() != null){
441449
if (i > 0 && data.binarySeparator.length > 0) {
442450
data.writer.write(data.binarySeparator);
443451
}
444452

445-
String[] field = data.dbFields.get(i);
446-
IValueMeta v;
453+
IValueMeta v = null;
454+
String streamFieldName = meta.getFields().get(i).getStreamField();
455+
String[] rowFields = data.outputRowMeta.getFieldNames();
456+
String streamFieldType = "";
457+
int streamIndex = -1;
458+
for(int j=0; j < rowFields.length ; j++){
459+
if(streamFieldName.equals(rowFields[j])){
460+
v = rowMeta.getValueMeta(j);
461+
streamIndex = j;
462+
}
463+
}
447464

448-
if (field[1].toUpperCase().startsWith("TIMESTAMP")) {
449-
v = new ValueMetaDate();
450-
v.setConversionMask("yyyy-MM-dd HH:mm:ss.SSS");
451-
} else if (field[1].toUpperCase().startsWith("DATE")) {
465+
boolean needConversion = false;
466+
if (v.getType() == IValueMeta.TYPE_TIMESTAMP) {
452467
v = new ValueMetaDate();
453-
v.setConversionMask("yyyy-MM-dd");
454-
} else if (field[1].toUpperCase().startsWith("TIME")) {
468+
v.setConversionMask("yyyy/MM/dd HH:mm:ss");
469+
needConversion = true;
470+
} else if (v.getType() == IValueMeta.TYPE_DATE) {
455471
v = new ValueMetaDate();
456-
v.setConversionMask("HH:mm:ss.SSS");
457-
} else if (field[1].toUpperCase().startsWith("NUMBER")
458-
|| field[1].toUpperCase().startsWith("FLOAT")) {
459-
v = new ValueMetaBigNumber();
460-
} else {
461-
v = new ValueMetaString();
462-
v.setLength(-1);
472+
v.setConversionMask("yyyy/MM/dd");
473+
needConversion = true;
463474
}
464475

465-
int fieldIndex = -1;
466-
if (data.fieldnrs.get(data.dbFields.get(i)[0]) != null) {
467-
fieldIndex = data.fieldnrs.get(data.dbFields.get(i)[0]);
468-
}
469476
Object valueData = null;
470-
if (fieldIndex >= 0) {
471-
valueData = v.convertData(rowMeta.getValueMeta(fieldIndex), row[fieldIndex]);
477+
if(streamIndex >= 0){
478+
if(needConversion){
479+
IValueMeta valueMeta = rowMeta.getValueMeta(streamIndex);
480+
Object obj = row[i];
481+
valueData = v.convertData(valueMeta, obj);
482+
}else{
483+
valueData = row[i];
484+
}
472485
} else if (meta.isErrorColumnMismatch()) {
473486
throw new HopException(
474-
"Error column mismatch: Database field "
475-
+ data.dbFields.get(i)[0]
487+
"Error column mismatch: Database streamField "
488+
+ meta.getFields().get(i).getStreamField()
476489
+ " not found on stream.");
477490
}
478491
writeField(v, valueData, data.binaryNullValue);

‎plugins/tech/aws/src/main/java/org/apache/hop/pipeline/transforms/redshift/bulkloader/RedshiftBulkLoaderDialog.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -767,9 +767,12 @@ private void generateMappings() {
767767

768768
// refresh data
769769
input.setTablename(variables.resolve(wTable.getText()));
770-
ITransformMeta transformMetaInterface = transformMeta.getTransform();
770+
ITransformMeta transformMeta = this.transformMeta.getTransform();
771+
if(StringUtils.isEmpty(input.getConnection())){
772+
input.setConnection(wConnection.getText());
773+
}
771774
try {
772-
targetFields = transformMetaInterface.getRequiredFields(variables);
775+
targetFields = transformMeta.getRequiredFields(variables);
773776
} catch (HopException e) {
774777
new ErrorDialog(
775778
shell,

‎plugins/transforms/excel/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
<name>Hop Plugins Transforms Excel</name>
3333

3434
<properties>
35-
<poi.version>5.2.3</poi.version>
35+
<poi.version>5.2.4</poi.version>
3636
<xerces.version>2.12.2</xerces.version>
3737
<odfdom-java.version>0.8.7</odfdom-java.version>
3838
</properties>

0 commit comments

Comments
 (0)