private void createScanRangeLocations(ParamCreateContext context, List<TBrokerFileStatus> fileStatuses)
            throws UserException {
        Preconditions.checkState(locationsHeap.isEmpty(), "Locations heap is not empty");

        long totalBytes = 0;
        for (TBrokerFileStatus fileStatus : fileStatuses) {
            totalBytes += fileStatus.size;
        }
        long numInstances = bytesPerInstance == 0 ? 1 : (totalBytes + bytesPerInstance - 1) / bytesPerInstance;
        // totalBytes may be 0, so numInstances may be 0
        numInstances = Math.max(numInstances, (long) 1);

        for (int i = 0; i < numInstances; ++i) {
            locationsHeap.add(Pair.create(newLocations(context.params, brokerDesc.getName(), brokerDesc.hasBroker()), 0L));
        }
    }
@Test
    public void testCreateScanRangeLocations(@Mocked GlobalStateMgr globalStateMgr,
                                             @Mocked SystemInfoService systemInfoService,
                                             @Injectable Database db, @Injectable OlapTable table)
            throws UserException {
        // table schema
        List<Column> columns = Lists.newArrayList();
        Column c1 = new Column("c1", Type.BIGINT, true);
        columns.add(c1);
        Column c2 = new Column("c2", Type.BIGINT, true);
        columns.add(c2);
        List<String> columnNames = Lists.newArrayList("c1", "c2");

        new Expectations() {
            {
                GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo();
                result = systemInfoService;
                systemInfoService.getIdToBackend();
                result = idToBackend;
                table.getBaseSchema();
                result = columns;
                table.getFullSchema();
                result = columns;
                table.getPartitions();
                minTimes = 0;
                result = Arrays.asList(partition);
                partition.getId();
                minTimes = 0;
                result = 0;
                table.getColumn("c1");
                result = columns.get(0);
                table.getColumn("c2");
                result = columns.get(1);
            }
        };

        // case 0
        // 2 csv files: file1 512M+, file2 256M-
        // result: 3 ranges. file1 3 ranges, file2 1 range

        // file groups
        List<BrokerFileGroup> fileGroups = Lists.newArrayList();
        List<String> files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1", "hdfs://127.0.0.1:9001/file2");
        DataDescription desc =
                new DataDescription("testTable", null, files, columnNames, null, null, null, false, null);
        BrokerFileGroup brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        fileGroups.add(brokerFileGroup);

        // file status
        List<List<TBrokerFileStatus>> fileStatusesList = Lists.newArrayList();
        List<TBrokerFileStatus> fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 536870968, true));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file2", false, 268435400, true));
        fileStatusesList.add(fileStatusList);

        Analyzer analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        DescriptorTable descTable = analyzer.getDescTbl();
        TupleDescriptor tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        FileScanNode scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList,
                2, WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        List<TScanRangeLocations> locationsList = scanNode.getScanRangeLocations(0);
        System.out.println(locationsList);
        Assert.assertEquals(3, locationsList.size());
        int file1RangesNum = 0;
        int file2RangesNum = 0;
        Set<Long> file1StartOffsetResult = Sets.newHashSet();
        Set<Long> file1RangeSizeResult = Sets.newHashSet();
        for (TScanRangeLocations locations : locationsList) {
            for (TBrokerRangeDesc rangeDesc : locations.scan_range.broker_scan_range.ranges) {
                long start = rangeDesc.start_offset;
                long size = rangeDesc.size;
                if (rangeDesc.path.endsWith("file1")) {
                    ++file1RangesNum;
                    file1StartOffsetResult.add(start);
                    file1RangeSizeResult.add(size);
                } else if (rangeDesc.path.endsWith("file2")) {
                    ++file2RangesNum;
                    Assert.assertTrue(start == 0);
                    Assert.assertTrue(size == 268435400);
                }
            }
        }
        Assert.assertEquals(Sets.newHashSet(0L, 268435456L, 536870912L), file1StartOffsetResult);
        Assert.assertEquals(Sets.newHashSet(56L, 268435456L), file1RangeSizeResult);
        Assert.assertEquals(3, file1RangesNum);
        Assert.assertEquals(1, file2RangesNum);

        // case 1
        // 4 parquet files
        // result: 3 ranges. 2 files in one range and 1 file in every other range

        // file groups
        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1", "hdfs://127.0.0.1:9001/file2",
                "hdfs://127.0.0.1:9001/file3", "hdfs://127.0.0.1:9001/file4");
        desc = new DataDescription("testTable", null, files, columnNames, null, null, "parquet", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "parquet");
        fileGroups.add(brokerFileGroup);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 268435454, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file2", false, 268435453, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file3", false, 268435452, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file4", false, 268435451, false));
        fileStatusesList.add(fileStatusList);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 4,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        Assert.assertEquals(3, locationsList.size());
        for (TScanRangeLocations locations : locationsList) {
            List<TBrokerRangeDesc> rangeDescs = locations.scan_range.broker_scan_range.ranges;
            Assert.assertTrue(rangeDescs.size() == 1 || rangeDescs.size() == 2);
        }

        // case 2
        // 2 file groups
        // result: 4 ranges. group1 3 ranges, group2 1 range

        // file groups
        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1", "hdfs://127.0.0.1:9001/file2",
                "hdfs://127.0.0.1:9001/file3");
        desc = new DataDescription("testTable", null, files, columnNames, null, null, "parquet", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "parquet");
        fileGroups.add(brokerFileGroup);

        List<String> files2 = Lists.newArrayList("hdfs://127.0.0.1:9001/file4", "hdfs://127.0.0.1:9001/file5");
        DataDescription desc2 =
                new DataDescription("testTable", null, files2, columnNames, null, null, null, false, null);
        BrokerFileGroup brokerFileGroup2 = new BrokerFileGroup(desc2);
        Deencapsulation.setField(brokerFileGroup2, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup2, "rowDelimiter", "\n");
        fileGroups.add(brokerFileGroup2);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 268435456, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file2", false, 10, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file3", false, 10, false));
        fileStatusesList.add(fileStatusList);

        List<TBrokerFileStatus> fileStatusList2 = Lists.newArrayList();
        fileStatusList2.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file4", false, 10, true));
        fileStatusList2.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file5", false, 10, true));
        fileStatusesList.add(fileStatusList2);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 5,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        Assert.assertEquals(4, locationsList.size());
        int group1RangesNum = 0;
        int group2RangesNum = 0;
        for (TScanRangeLocations locations : locationsList) {
            List<TBrokerRangeDesc> rangeDescs = locations.scan_range.broker_scan_range.ranges;
            String path = rangeDescs.get(0).path;
            if (path.endsWith("file1") || path.endsWith("file2") || path.endsWith("file3")) {
                Assert.assertEquals(1, rangeDescs.size());
                ++group1RangesNum;
            } else if (path.endsWith("file4") || path.endsWith("file5")) {
                Assert.assertEquals(2, rangeDescs.size());
                ++group2RangesNum;
            }
        }
        Assert.assertEquals(3, group1RangesNum);
        Assert.assertEquals(1, group2RangesNum);

        // case 4
        // 2 parquet file and one is very large
        // result: 2 ranges

        // file groups
        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1", "hdfs://127.0.0.1:9001/file2");
        desc = new DataDescription("testTable", null, files, columnNames, null, null, "parquet", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "parquet");
        fileGroups.add(brokerFileGroup);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 268435456000L, false));
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file2", false, 10, false));
        fileStatusesList.add(fileStatusList);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 2,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        System.out.println(locationsList);
        Assert.assertEquals(2, locationsList.size());

        // case 5
        // 1 file which size is 0
        // result: 1 range

        // file groups
        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1");
        desc = new DataDescription("testTable", null, files, columnNames, null, null, "parquet", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "parquet");
        fileGroups.add(brokerFileGroup);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 0, false));
        fileStatusesList.add(fileStatusList);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 1,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        System.out.println(locationsList);
        Assert.assertEquals(1, locationsList.size());
        List<TBrokerRangeDesc> rangeDescs = locationsList.get(0).scan_range.broker_scan_range.ranges;
        Assert.assertEquals(1, rangeDescs.size());
        Assert.assertEquals(0, rangeDescs.get(0).size);

        // case 5
        // 1 file which size is 0 in json format
        // result: 1 range
        // file groups

        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1");
        desc = new DataDescription("testTable", null, files, columnNames, null, null, "json", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "json");
        fileGroups.add(brokerFileGroup);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        fileStatusList.add(new TBrokerFileStatus("hdfs://127.0.0.1:9001/file1", false, 0, false));
        fileStatusesList.add(fileStatusList);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 1,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        System.out.println(locationsList);
        Assert.assertEquals(1, locationsList.size());
        rangeDescs = locationsList.get(0).scan_range.broker_scan_range.ranges;
        Assert.assertEquals(1, rangeDescs.size());
        Assert.assertEquals(0, rangeDescs.get(0).size);

        // case 6
        // csv file compression type
        // result: CSV_PLAIN, CSV_GZ, CSV_BZ2, CSV_LZ4, CSV_DFLATE, CSV_ZSTD

        // file groups
        fileGroups = Lists.newArrayList();
        files = Lists.newArrayList("hdfs://127.0.0.1:9001/file1", "hdfs://127.0.0.1:9001/file2.csv",
                "hdfs://127.0.0.1:9001/file3.gz", "hdfs://127.0.0.1:9001/file4.bz2", "hdfs://127.0.0.1:9001/file5.lz4",
                "hdfs://127.0.0.1:9001/file6.deflate", "hdfs://127.0.0.1:9001/file7.zst");
        desc =
                new DataDescription("testTable", null, files, columnNames, null, null, "csv", false, null);
        brokerFileGroup = new BrokerFileGroup(desc);
        Deencapsulation.setField(brokerFileGroup, "columnSeparator", "\t");
        Deencapsulation.setField(brokerFileGroup, "rowDelimiter", "\n");
        Deencapsulation.setField(brokerFileGroup, "fileFormat", "csv");
        fileGroups.add(brokerFileGroup);

        // file status
        fileStatusesList = Lists.newArrayList();
        fileStatusList = Lists.newArrayList();
        for (String file : files) {
            fileStatusList.add(new TBrokerFileStatus(file, false, 1024, true));
        }
        fileStatusesList.add(fileStatusList);

        analyzer = new Analyzer(GlobalStateMgr.getCurrentState(), new ConnectContext());
        descTable = analyzer.getDescTbl();
        tupleDesc = descTable.createTupleDescriptor("DestTableTuple");
        scanNode = new FileScanNode(new PlanNodeId(0), tupleDesc, "FileScanNode", fileStatusesList, 2,
                WarehouseManager.DEFAULT_WAREHOUSE_ID);
        scanNode.setLoadInfo(jobId, txnId, table, brokerDesc, fileGroups, true, loadParallelInstanceNum);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);

        // check
        locationsList = scanNode.getScanRangeLocations(0);
        Assert.assertEquals(1, locationsList.size());

        Assert.assertEquals(7, locationsList.get(0).scan_range.broker_scan_range.ranges.size());

        Assert.assertEquals(TFileFormatType.FORMAT_CSV_PLAIN,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(0).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_PLAIN,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(1).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_GZ,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(2).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_BZ2,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(3).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_LZ4_FRAME,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(4).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_DEFLATE,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(5).format_type);
        Assert.assertEquals(TFileFormatType.FORMAT_CSV_ZSTD,
                locationsList.get(0).scan_range.broker_scan_range.ranges.get(6).format_type);
    }