@Override
    public void init() throws LoadException {
        createEtlJobConf();
    }
@Test
    public void testRangePartitionHashDistribution(@Injectable SparkLoadJob sparkLoadJob,
                                                   @Injectable SparkResource resource,
                                                   @Injectable BrokerDesc brokerDesc,
                                                   @Mocked GlobalStateMgr globalStateMgr,
                                                   @Injectable Database database,
                                                   @Injectable OlapTable table)
            throws LoadException, DdlException, AnalysisException {
        long dbId = 0L;
        long tableId = 1L;

        // c1 is partition column, c2 is distribution column
        List<Column> columns = Lists.newArrayList();
        columns.add(new Column("c1", Type.INT, true, null, false, null, ""));
        columns.add(new Column("c2", ScalarType.createVarchar(10), true, null, false, null, ""));
        columns.add(new Column("c3", Type.INT, false, AggregateType.SUM, false, null, ""));

        Map<ColumnId, Column> idToColumn = Maps.newTreeMap(ColumnId.CASE_INSENSITIVE_ORDER);
        for (Column column : columns) {
            idToColumn.put(column.getColumnId(), column);
        }

        // indexes
        Map<Long, List<Column>> indexIdToSchema = Maps.newHashMap();
        long index1Id = 3L;
        indexIdToSchema.put(index1Id, columns);
        long index2Id = 4L;
        indexIdToSchema.put(index2Id, Lists.newArrayList(columns.get(0), columns.get(2)));

        // partition and distribution info
        long partition1Id = 2L;
        long partition2Id = 5L;
        // partition3 is temporary partition
        long partition3Id = 6L;
        int distributionColumnIndex = 1;
        DistributionInfo distributionInfo =
                new HashDistributionInfo(3, Lists.newArrayList(columns.get(distributionColumnIndex)));
        Partition partition1 = new Partition(partition1Id, "p1", null,
                distributionInfo);
        Partition partition2 = new Partition(partition2Id, "p2", null,
                new HashDistributionInfo(4, Lists.newArrayList(columns.get(distributionColumnIndex))));
        Partition partition3 = new Partition(partition3Id, "tp3", null,
                distributionInfo);
        int partitionColumnIndex = 0;
        List<Partition> partitions = Lists.newArrayList(partition1, partition2);
        RangePartitionInfo partitionInfo =
                new RangePartitionInfo(Lists.newArrayList(columns.get(partitionColumnIndex)));
        PartitionKeyDesc partitionKeyDesc1 = new PartitionKeyDesc(Lists.newArrayList(new PartitionValue("10")));
        SingleRangePartitionDesc partitionDesc1 = new SingleRangePartitionDesc(false, "p1", partitionKeyDesc1, null);
        partitionDesc1.analyze(1, null);
        partitionInfo.handleNewSinglePartitionDesc(MetaUtils.buildIdToColumn(columns),
                partitionDesc1, partition1Id, false);
        PartitionKeyDesc partitionKeyDesc2 = new PartitionKeyDesc(Lists.newArrayList(new PartitionValue("20")));
        SingleRangePartitionDesc partitionDesc2 = new SingleRangePartitionDesc(false, "p2", partitionKeyDesc2, null);
        partitionDesc2.analyze(1, null);
        partitionInfo.handleNewSinglePartitionDesc(MetaUtils.buildIdToColumn(columns),
                partitionDesc2, partition2Id, false);
        PartitionKeyDesc partitionKeyDesc3 = new PartitionKeyDesc(Lists.newArrayList(new PartitionValue("10")));
        SingleRangePartitionDesc partitionDesc3 = new SingleRangePartitionDesc(false, "tp3", partitionKeyDesc1, null);
        partitionDesc3.analyze(1, null);
        partitionInfo.handleNewSinglePartitionDesc(MetaUtils.buildIdToColumn(columns),
                partitionDesc3, partition3Id, true);

        new Expectations() {
            {
                globalStateMgr.getDb(dbId);
                result = database;
                database.getTable(tableId);
                result = table;
                table.getPartitions();
                result = partitions;
                table.getIndexIdToSchema();
                result = indexIdToSchema;
                table.getDefaultDistributionInfo();
                result = distributionInfo;
                table.getSchemaHashByIndexId(index1Id);
                result = 123;
                table.getSchemaHashByIndexId(index2Id);
                result = 234;
                table.getPartitionInfo();
                result = partitionInfo;
                table.getPartition(partition1Id);
                result = partition1;
                table.getPartition(partition2Id);
                result = partition2;
                table.getPartition(partition3Id);
                result = partition3;
                table.getKeysTypeByIndexId(index1Id);
                result = KeysType.AGG_KEYS;
                table.getKeysTypeByIndexId(index2Id);
                result = KeysType.AGG_KEYS;
                table.getBaseIndexId();
                result = index1Id;
                table.getIdToColumn();
                result = idToColumn;
            }
        };

        // case 0: partition is null in load stmt
        // file group
        Map<BrokerFileGroupAggInfo.FileGroupAggKey, List<BrokerFileGroup>> aggKeyToFileGroups = Maps.newHashMap();
        List<BrokerFileGroup> brokerFileGroups = Lists.newArrayList();
        DataDescription desc = new DataDescription("testTable", null, Lists.newArrayList("abc.txt"),
                null, null, null, null, false, null);
        BrokerFileGroup brokerFileGroup = new BrokerFileGroup(desc);
        brokerFileGroups.add(brokerFileGroup);
        BrokerFileGroupAggInfo.FileGroupAggKey aggKey = new BrokerFileGroupAggInfo.FileGroupAggKey(tableId, null);
        aggKeyToFileGroups.put(aggKey, brokerFileGroups);

        // create pending task
        SparkLoadPendingTask task = new SparkLoadPendingTask(sparkLoadJob, aggKeyToFileGroups, resource, brokerDesc);
        EtlJobConfig etlJobConfig = Deencapsulation.getField(task, "etlJobConfig");
        Assert.assertEquals(null, etlJobConfig);
        task.init();
        etlJobConfig = Deencapsulation.getField(task, "etlJobConfig");
        Assert.assertTrue(etlJobConfig != null);

        // check table id
        Map<Long, EtlTable> idToEtlTable = etlJobConfig.tables;
        Assert.assertEquals(1, idToEtlTable.size());
        Assert.assertTrue(idToEtlTable.containsKey(tableId));

        // check indexes
        EtlTable etlTable = idToEtlTable.get(tableId);
        List<EtlIndex> etlIndexes = etlTable.indexes;
        Assert.assertEquals(2, etlIndexes.size());
        Assert.assertEquals(index1Id, etlIndexes.get(0).indexId);
        Assert.assertEquals(index2Id, etlIndexes.get(1).indexId);

        // check base index columns
        EtlIndex baseIndex = etlIndexes.get(0);
        Assert.assertTrue(baseIndex.isBaseIndex);
        Assert.assertEquals(3, baseIndex.columns.size());
        for (int i = 0; i < columns.size(); i++) {
            Assert.assertEquals(columns.get(i).getName(), baseIndex.columns.get(i).columnName);
        }
        Assert.assertEquals("AGGREGATE", baseIndex.indexType);

        // check partitions
        EtlPartitionInfo etlPartitionInfo = etlTable.partitionInfo;
        Assert.assertEquals("RANGE", etlPartitionInfo.partitionType);
        List<String> partitionColumns = etlPartitionInfo.partitionColumnRefs;
        Assert.assertEquals(1, partitionColumns.size());
        Assert.assertEquals(columns.get(partitionColumnIndex).getName(), partitionColumns.get(0));
        List<String> distributionColumns = etlPartitionInfo.distributionColumnRefs;
        Assert.assertEquals(1, distributionColumns.size());
        Assert.assertEquals(columns.get(distributionColumnIndex).getName(), distributionColumns.get(0));
        List<EtlPartition> etlPartitions = etlPartitionInfo.partitions;
        Assert.assertEquals(2, etlPartitions.size());
        Assert.assertEquals(partition1Id, etlPartitions.get(0).partitionId);
        Assert.assertEquals(partition2Id, etlPartitions.get(1).partitionId);

        // check file group
        List<EtlFileGroup> etlFileGroups = etlTable.fileGroups;
        Assert.assertEquals(1, etlFileGroups.size());

        // case 1: temporary partition in load stmt
        // file group
        aggKeyToFileGroups = Maps.newHashMap();
        brokerFileGroups = Lists.newArrayList();
        PartitionNames partitionNames = new PartitionNames(true, Lists.newArrayList("tp3"));
        desc = new DataDescription("testTable", partitionNames, Lists.newArrayList("abc.txt"),
                null, null, null, null, false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        brokerFileGroups.add(brokerFileGroup);
        aggKey = new BrokerFileGroupAggInfo.FileGroupAggKey(tableId, Lists.newArrayList(partition3Id));
        aggKeyToFileGroups.put(aggKey, brokerFileGroups);

        // create pending task
        task = new SparkLoadPendingTask(sparkLoadJob, aggKeyToFileGroups, resource, brokerDesc);
        task.init();

        etlJobConfig = Deencapsulation.getField(task, "etlJobConfig");
        Assert.assertTrue(etlJobConfig != null);
        idToEtlTable = etlJobConfig.tables;
        etlTable = idToEtlTable.get(tableId);

        // check partitions
        etlPartitionInfo = etlTable.partitionInfo;
        Assert.assertEquals("RANGE", etlPartitionInfo.partitionType);
        partitionColumns = etlPartitionInfo.partitionColumnRefs;
        Assert.assertEquals(1, partitionColumns.size());
        Assert.assertEquals(columns.get(partitionColumnIndex).getName(), partitionColumns.get(0));
        distributionColumns = etlPartitionInfo.distributionColumnRefs;
        Assert.assertEquals(1, distributionColumns.size());
        Assert.assertEquals(columns.get(distributionColumnIndex).getName(), distributionColumns.get(0));
        etlPartitions = etlPartitionInfo.partitions;
        Assert.assertEquals(1, etlPartitions.size());
        Assert.assertEquals(partition3Id, etlPartitions.get(0).partitionId);
    }