Skip to content

Commit 12a93e5

Browse files
deniskuzZdeniskuzZ
committed
refactor #4 
1 parent e80df54 commit 12a93e5

File tree

8 files changed

+23
-162
lines changed

8 files changed

+23
-162
lines changed

‎accumulo-handler/src/java/org/apache/hadoop/hive/accumulo/AccumuloStorageHandler.java

Lines changed: 1 addition & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@
3838
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
3939
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
4040
import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
41-
import org.apache.hadoop.hive.ql.metadata.HiveException;
4241
import org.apache.hadoop.hive.ql.metadata.HiveStoragePredicateHandler;
4342
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
4443
import org.apache.hadoop.hive.ql.plan.TableDesc;
@@ -72,7 +71,6 @@ public class AccumuloStorageHandler extends DefaultStorageHandler implements Hiv
7271

7372
protected AccumuloPredicateHandler predicateHandler = AccumuloPredicateHandler.getInstance();
7473
protected AccumuloConnectionParameters connectionParams;
75-
protected Configuration conf;
7674
protected HiveAccumuloHelper helper = new HiveAccumuloHelper();
7775

7876
/**
@@ -146,23 +144,12 @@ protected String getTableName(TableDesc tableDesc) {
146144
return tableName;
147145
}
148146

149-
protected String getColumnTypes(TableDesc tableDesc) {
150-
Properties props = tableDesc.getProperties();
151-
String columnTypes = props.getProperty(serdeConstants.LIST_COLUMN_TYPES);
152-
return columnTypes;
153-
}
154-
155147
protected String getIndexTableName(TableDesc tableDesc) {
156148
Properties props = tableDesc.getProperties();
157149
String tableName = props.getProperty(AccumuloIndexParameters.INDEXTABLE_NAME);
158150
return tableName;
159151
}
160152

161-
@Override
162-
public Configuration getConf() {
163-
return conf;
164-
}
165-
166153
@Override
167154
public void setConf(Configuration conf) {
168155
this.conf = conf;
@@ -180,7 +167,7 @@ public HiveMetaHook getMetaHook() {
180167
}
181168

182169
@Override
183-
public HiveAuthorizationProvider getAuthorizationProvider() throws HiveException {
170+
public HiveAuthorizationProvider getAuthorizationProvider() {
184171
return null;
185172
}
186173

@@ -294,7 +281,6 @@ private String getIndexedColFamQuals(TableDesc tableDesc, String indexColumns, S
294281

295282
String cols = indexColumns;
296283

297-
298284
String hiveColString = tableDesc.getProperties().getProperty(serdeConstants.LIST_COLUMNS);
299285
// if there are actual accumulo index columns defined then build
300286
// the comma separated list of accumulo columns

‎hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FosterStorageHandler.java

Lines changed: 1 addition & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -19,28 +19,20 @@
1919

2020
package org.apache.hive.hcatalog.mapreduce;
2121

22-
import org.apache.hadoop.conf.Configuration;
2322
import org.apache.hadoop.fs.Path;
2423
import org.apache.hadoop.hive.common.FileUtils;
2524
import org.apache.hadoop.hive.common.JavaUtils;
26-
import org.apache.hadoop.hive.metastore.HiveMetaHook;
2725
import org.apache.hadoop.hive.ql.io.AcidUtils;
2826
import org.apache.hadoop.hive.ql.io.IOConstants;
29-
import org.apache.hadoop.hive.ql.metadata.HiveException;
3027
import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
3128
import org.apache.hadoop.hive.ql.plan.TableDesc;
32-
import org.apache.hadoop.hive.ql.security.authorization.DefaultHiveAuthorizationProvider;
33-
import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
3429
import org.apache.hadoop.hive.serde2.AbstractSerDe;
3530
import org.apache.hadoop.mapred.InputFormat;
36-
import org.apache.hadoop.mapred.JobConf;
3731
import org.apache.hadoop.mapred.OutputFormat;
3832
import org.apache.hive.hcatalog.common.HCatConstants;
3933
import org.apache.hive.hcatalog.common.HCatUtil;
4034
import org.apache.hive.hcatalog.data.schema.HCatFieldSchema;
4135
import org.apache.hive.hcatalog.data.schema.HCatSchema;
42-
import org.slf4j.Logger;
43-
import org.slf4j.LoggerFactory;
4436

4537
import java.io.IOException;
4638
import java.util.ArrayList;
@@ -55,12 +47,6 @@
5547
*/
5648
public class FosterStorageHandler extends DefaultStorageHandler {
5749

58-
private static final Logger LOG = LoggerFactory.getLogger(FosterStorageHandler.class);
59-
60-
public Configuration conf;
61-
/** The directory under which data is initially written for a non partitioned table */
62-
protected static final String TEMP_DIR_NAME = "_TEMP";
63-
6450
private Class<? extends InputFormat> ifClass;
6551
private Class<? extends OutputFormat> ofClass;
6652
private Class<? extends AbstractSerDe> serDeClass;
@@ -94,16 +80,6 @@ public Class<? extends AbstractSerDe> getSerDeClass() {
9480
return serDeClass; //To change body of implemented methods use File | Settings | File Templates.
9581
}
9682

97-
@Override
98-
public HiveMetaHook getMetaHook() {
99-
return null;
100-
}
101-
102-
@Override
103-
public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
104-
//do nothing currently
105-
}
106-
10783
@Override
10884
public void configureInputJobProperties(TableDesc tableDesc,
10985
Map<String, String> jobProperties) {
@@ -148,7 +124,6 @@ public void configureInputJobProperties(TableDesc tableDesc,
148124
} catch (IOException e) {
149125
throw new IllegalStateException("Failed to set output path", e);
150126
}
151-
152127
}
153128

154129
@Override
@@ -224,39 +199,11 @@ public void configureOutputJobProperties(TableDesc tableDesc,
224199

225200
SpecialCases.addSpecialCasesParametersToOutputJobProperties(jobProperties, jobInfo, ofClass);
226201

227-
228202
jobProperties.put(HCatConstants.HCAT_KEY_OUTPUT_INFO,
229203
HCatUtil.serialize(jobInfo));
230204
} catch (IOException e) {
231205
throw new IllegalStateException("Failed to set output path", e);
232206
}
233-
234-
}
235-
236-
public void configureTableJobProperties(TableDesc tableDesc,
237-
Map<String, String> jobProperties) {
238-
return;
239-
}
240-
241-
OutputFormatContainer getOutputFormatContainer(
242-
org.apache.hadoop.mapred.OutputFormat outputFormat) {
243-
return new FileOutputFormatContainer(outputFormat);
244-
}
245-
246-
@Override
247-
public Configuration getConf() {
248-
return conf;
249-
}
250-
251-
@Override
252-
public void setConf(Configuration conf) {
253-
this.conf = conf;
254-
}
255-
256-
@Override
257-
public HiveAuthorizationProvider getAuthorizationProvider()
258-
throws HiveException {
259-
return new DefaultHiveAuthorizationProvider();
260207
}
261208

262-
}
209+
}

‎iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

Lines changed: 10 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@
8787
import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader;
8888
import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
8989
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
90+
import org.apache.hadoop.hive.ql.metadata.DefaultStorageHandler;
9091
import org.apache.hadoop.hive.ql.metadata.DummyPartition;
9192
import org.apache.hadoop.hive.ql.metadata.HiveException;
9293
import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
@@ -222,7 +223,7 @@
222223
import static org.apache.iceberg.TableProperties.MERGE_MODE;
223224
import static org.apache.iceberg.TableProperties.UPDATE_MODE;
224225

225-
public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, HiveStorageHandler {
226+
public class HiveIcebergStorageHandler extends DefaultStorageHandler implements HiveStoragePredicateHandler {
226227
private static final Logger LOG = LoggerFactory.getLogger(HiveIcebergStorageHandler.class);
227228

228229
private static final String ICEBERG_URI_PREFIX = "iceberg://";
@@ -245,8 +246,6 @@ public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, H
245246

246247
private static final List<FieldSchema> EMPTY_ORDERING = ImmutableList.of();
247248

248-
private Configuration conf;
249-
250249
@Override
251250
public Class<? extends InputFormat> getInputFormatClass() {
252251
return HiveIcebergInputFormat.class;
@@ -316,17 +315,6 @@ public void commitJob(JobContext originalContext) {
316315
}
317316
}
318317

319-
@Override
320-
public void configureTableJobProperties(TableDesc tableDesc, Map<String, String> map) {
321-
322-
}
323-
324-
// Override annotation commented out, since this interface method has been introduced only in Hive 3
325-
// @Override
326-
public void configureInputJobCredentials(TableDesc tableDesc, Map<String, String> secrets) {
327-
328-
}
329-
330318
@Override
331319
public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
332320
setCommonJobConf(jobConf);
@@ -378,21 +366,6 @@ public boolean supportsPartitioning() {
378366
return true;
379367
}
380368

381-
@Override
382-
public Configuration getConf() {
383-
return conf;
384-
}
385-
386-
@Override
387-
public void setConf(Configuration conf) {
388-
this.conf = conf;
389-
}
390-
391-
@Override
392-
public String toString() {
393-
return this.getClass().getName();
394-
}
395-
396369
/**
397370
* @param jobConf Job configuration for InputFormat to access
398371
* @param deserializer Deserializer
@@ -1357,11 +1330,12 @@ static String encodeString(String rawString) {
13571330
return HiveConf.EncoderDecoderFactory.URL_ENCODER_DECODER.encode(rawString);
13581331
}
13591332

1360-
String getPathForAuth(String locationProperty) {
1333+
private String getPathForAuth(String locationProperty) {
13611334
return getPathForAuth(locationProperty,
13621335
SessionStateUtil.getProperty(conf, SessionStateUtil.DEFAULT_TABLE_LOCATION).orElse(null));
13631336
}
13641337

1338+
@VisibleForTesting
13651339
String getPathForAuth(String locationProperty, String defaultTableLocation) {
13661340
boolean maskDefaultLocation = conf.getBoolean(ConfVars.HIVE_ICEBERG_MASK_DEFAULT_LOCATION.varname,
13671341
ConfVars.HIVE_ICEBERG_MASK_DEFAULT_LOCATION.defaultBoolVal);
@@ -1393,7 +1367,6 @@ private boolean arePathsInSameFs(String locationProperty, String defaultTableLoc
13931367

13941368
@Override
13951369
public void validateSinkDesc(FileSinkDesc sinkDesc) throws SemanticException {
1396-
HiveStorageHandler.super.validateSinkDesc(sinkDesc);
13971370
if (sinkDesc.getInsertOverwrite()) {
13981371
Table table = IcebergTableUtil.getTable(conf, sinkDesc.getTableInfo().getProperties());
13991372
if (table.currentSnapshot() != null &&
@@ -1486,7 +1459,7 @@ public List<FieldSchema> sortColumns(org.apache.hadoop.hive.ql.metadata.Table hm
14861459
s.transform().toString(), s.direction().name(), s.nullOrder().name()))).collect(Collectors.toList());
14871460
}
14881461

1489-
private void setCommonJobConf(JobConf jobConf) {
1462+
private static void setCommonJobConf(JobConf jobConf) {
14901463
jobConf.set("tez.mrreader.config.update.properties", "hive.io.file.readcolumn.names,hive.io.file.readcolumn.ids");
14911464
}
14921465

@@ -1642,12 +1615,13 @@ static void overlayTableProperties(Configuration configuration, TableDesc tableD
16421615

16431616
// serialize table object into config
16441617
Table serializableTable = SerializableTable.copyOf(table);
1618+
16451619
// set table format-version and write-mode information from tableDesc
1646-
List<String> writeConfigList = ImmutableList.of(
1620+
final List<String> writeConfigList = ImmutableList.of(
16471621
FORMAT_VERSION, DELETE_MODE, UPDATE_MODE, MERGE_MODE);
1648-
if (IcebergTableUtil.isV2TableOrAbove(props::getProperty)) {
1649-
writeConfigList.forEach(cfg -> serializableTable.properties().computeIfAbsent(cfg, props::getProperty));
1650-
}
1622+
writeConfigList.forEach(cfg ->
1623+
serializableTable.properties().computeIfAbsent(cfg, props::getProperty));
1624+
16511625
checkAndSkipIoConfigSerialization(configuration, serializableTable);
16521626
map.put(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableDesc.getTableName(),
16531627
SerializationUtil.serializeToBase64(serializableTable));
@@ -2038,12 +2012,6 @@ public void validatePartSpec(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
20382012
Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
20392013
List<PartitionField> partitionFields = IcebergTableUtil.getPartitionFields(table,
20402014
policy != RewritePolicy.PARTITION);
2041-
validatePartSpecImpl(hmsTable, partitionSpec, partitionFields);
2042-
}
2043-
2044-
private void validatePartSpecImpl(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
2045-
Map<String, String> partitionSpec, List<PartitionField> partitionFields) throws SemanticException {
2046-
Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
20472015
if (hmsTable.getSnapshotRef() != null && IcebergTableUtil.hasUndergonePartitionEvolution(table)) {
20482016
// for this case we rewrite the query as delete query, so validations would be done as part of delete.
20492017
return;
@@ -2139,11 +2107,6 @@ public boolean isPartitioned(org.apache.hadoop.hive.ql.metadata.Table hmsTable)
21392107
public Partition getPartition(org.apache.hadoop.hive.ql.metadata.Table table,
21402108
Map<String, String> partitionSpec, RewritePolicy policy) throws SemanticException {
21412109
validatePartSpec(table, partitionSpec, policy);
2142-
return getPartitionImpl(table, partitionSpec);
2143-
}
2144-
2145-
private Partition getPartitionImpl(org.apache.hadoop.hive.ql.metadata.Table table,
2146-
Map<String, String> partitionSpec) throws SemanticException {
21472110
try {
21482111
String partName = Warehouse.makePartName(partitionSpec, false);
21492112
return new DummyPartition(table, partName, partitionSpec);

‎iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,7 @@ private static void checkResiduals(ScanTaskGroup<FileScanTask> task) {
266266

267267
@Override
268268
public RecordReader<Void, T> createRecordReader(InputSplit split, TaskAttemptContext context) {
269-
return split instanceof IcebergMergeSplit ? new IcebergMergeRecordReader<>() : new IcebergRecordReader<T>();
269+
return split instanceof IcebergMergeSplit ?
270+
new IcebergMergeRecordReader<>() : new IcebergRecordReader<>();
270271
}
271272
}

‎iceberg/iceberg-handler/src/test/queries/positive/iceberg_v2_deletes.q

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ select * from ice01 order by id;
5959
insert into icepart01 values (1,1),(2,1),(3,2),(4,2);
6060

6161
-- check the inserted values
62-
select * from icepart01 order by id;;
62+
select * from icepart01 order by id;
6363

6464
-- delete some values
6565
delete from icepart01 where id>=2 AND id<4;
@@ -86,7 +86,7 @@ show create table icepart01;
8686
delete from icepart01 where id=5;
8787

8888
-- check the entries, the deleted entries shouldn't be there.
89-
select * from icepart01 order by id;;
89+
select * from icepart01 order by id;
9090

9191
-- clean up
9292
drop table ice01;

‎iceberg/iceberg-handler/src/test/queries/positive/iceberg_v3_deletion_vectors.q

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ select * from ice01 order by id;
5252
insert into icepart01 values (1,1),(2,1),(3,2),(4,2);
5353

5454
-- check the inserted values
55-
select * from icepart01 order by id;;
55+
select * from icepart01 order by id;
5656

5757
-- delete some values
5858
delete from icepart01 where id>=2 AND id<4;
@@ -72,7 +72,7 @@ delete from icepart01 where id=7;
7272
select content, file_format, spec_id, record_count, content_offset from default.icepart01.delete_files;
7373

7474
-- check the entries, the deleted entries shouldn't be there.
75-
select * from icepart01 order by id;;
75+
select * from icepart01 order by id;
7676

7777
-- clean up
7878
drop table ice01;

0 commit comments

Comments
 (0)