public static List<String> parseColumnsFromPath(String filePath, List<String> columnsFromPath)
            throws UserException {
        if (columnsFromPath == null || columnsFromPath.isEmpty()) {
            return Collections.emptyList();
        }
        String[] strings = filePath.split("/");
        if (strings.length < 2) {
            throw new UserException(
                    "Fail to parse columnsFromPath, expected: " + columnsFromPath + ", filePath: " + filePath);
        }
        String[] columns = new String[columnsFromPath.size()];
        for (int i = strings.length - 1; i >= 0; i--) {
            String str = strings[i];
            if (str == null || str.isEmpty() || !str.contains("=")) {
                continue;
            }
            String[] pair = str.split("=", 2);
            if (pair.length != 2) {
                continue;
            }
            int index = columnsFromPath.indexOf(pair[0]);
            if (index == -1) {
                continue;
            }
            columns[index] = pair[1];
        }

        for (int i = 0; i < columns.length; i++) {
            if (columns[i] == null) {
                throw new UserException(
                        "Fail to parse columnsFromPath, expected: " + columnsFromPath.get(i) + ", filePath: " + filePath);
            }
        }
        return Lists.newArrayList(columns);
    }
@Test
    public void testColumnFromPath() {
        try {
            // normal case
            List<String> columns = HdfsUtil.parseColumnsFromPath("hdfs://key1=val1/some_path/key2=val2/key3=val3/*", Arrays.asList("key3", "key2", "key1"));
            Assert.assertEquals(3, columns.size());
            Assert.assertEquals("val3", columns.get(0));
            Assert.assertEquals("val2", columns.get(1));
            Assert.assertEquals("val1", columns.get(2));

            // invalid path
            Assert.assertThrows(UserException.class, () ->
                    HdfsUtil.parseColumnsFromPath("invalid_path", Arrays.asList("key3", "key2", "key1")));

            // missing key of columns from path
            Assert.assertThrows(UserException.class, () ->
                    HdfsUtil.parseColumnsFromPath("hdfs://key1=val1/some_path/key3=val3/*", Arrays.asList("key3", "key2", "key1")));
        } catch (UserException e) {
            Assert.fail(e.getMessage());
        }
    }